use std::{
collections::HashMap,
path::{Path, PathBuf},
sync::Arc,
time::{Duration, Instant},
};
use wasmtime::{component::Linker, Store};
use wasmtime_wasi::{
DirPerms, FilePerms, ResourceTable, WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView,
};
use wasmtime_wasi_http::{
p2::{WasiHttpCtxView, WasiHttpView},
WasiHttpCtx,
};
use murmur_artifact::{
HookBinding, HookCommitPolicy, HookConfig, HookExecutionMode, HookOverflowPolicy,
PACKED_MANIFEST_ENTRY,
};
use tokio::sync::mpsc;
use crate::{
artifact_config::ARTIFACT_CONFIG_ENV,
bindings::hook::exports::murmur::hook::lifecycle::{
CompactionEvent, HookOutput, InferenceEvent, Message, SessionContext, SessionEndEvent,
ShellEvent, ShellOutcome, StageEvent, TaskEndEvent, TaskStartEvent, ToolEvent, ToolOutcome,
},
conversation_import::{add_conversation_to_linker, ConversationState},
errors::RuntimeError,
inference_import::{add_inference_to_linker, HookInferenceCtx, HookInferenceRecord},
limits::{classify_guest_failure, ExecutionLimiter, ExecutionLimits},
network_policy::{resolve_scoped_dir, HookCapabilityGrant},
runtime::NetworkPolicyHooks,
state_store::STATE_PREOPEN_NAME,
task_io_import::{add_task_io_to_linker, TaskIoState},
tokens_import::add_tokens_to_linker,
types::StagedHookArtifact,
};
const LIFECYCLE_IFACE: &str = "murmur:hook/lifecycle@0.8.0";
fn resolve_lifecycle_iface(
instance: &wasmtime::component::Instance,
store: &mut Store<HookStoreState>,
) -> Option<wasmtime::component::ComponentExportIndex> {
instance.get_export_index(&mut *store, None, LIFECYCLE_IFACE)
}
fn missing_lifecycle_msg(subject: &str) -> String {
format!(
"hook {subject} does not export {LIFECYCLE_IFACE}; rebuild the hook against the current WIT (run `mur install` for a default artifact, or rebuild from source otherwise)"
)
}
const ASYNC_HOOK_QUEUE_DEPTH: usize = 1024;
const ASYNC_HOOK_DRAIN_TIMEOUT: Duration = Duration::from_secs(15);
const FAULT_EVENT_DRAIN: &str = "drain";
const FAULT_ARM_ERROR: &str = "error";
const FAULT_ARM_QUEUE_OVERFLOW: &str = "queue-overflow";
const FAULT_ARM_TIMEOUT: &str = "timeout";
pub(crate) const FAULT_ARM_SEED_REJECTED: &str = "seed-rejected";
pub(crate) struct HookRuntime {
workdir: PathBuf,
blocking_hooks: Vec<HookInstance>,
async_hooks: Vec<AsyncHookQueue>,
fault_tx: mpsc::UnboundedSender<DispatchFault>,
fault_rx: mpsc::UnboundedReceiver<DispatchFault>,
context: SessionContextData,
inference: Option<Arc<HookInferenceCtx>>,
task_io: Arc<TaskIoState>,
conversation: Arc<ConversationState>,
dispatch_faults: Vec<DispatchFault>,
gates_calls: bool,
started: Instant,
total_input_tokens: u64,
total_output_tokens: u64,
total_tool_calls: u32,
total_shell_calls: u32,
total_turns: u32,
}
const REQUIRED_HOOK_FNS: [&str; 6] = [
"on-session-start",
"on-inference",
"on-tool-call",
"on-shell",
"on-compaction",
"on-session-end",
];
const OPTIONAL_HOOK_FNS: [&str; 2] = ["on-task-start", "on-task-end"];
const HONORED_OUTPUT_ARM: &[(&str, Option<&str>)] = &[
("on-stage", Some("write-manifests")),
("on-session-start", None),
("on-task-start", Some("seed-context")),
("on-inference", Some("artifact")),
("on-tool-call", None),
("on-shell", None),
("on-compaction", Some("replace-context")),
("on-task-end", Some("reopen-task")),
("on-session-end", None),
];
const HONORED_DECISION_ARM: &[(&str, Option<&str>)] =
&[("on-tool-call", Some("deny")), ("on-shell", Some("deny"))];
const SHELL_COMMAND_DISPLAY_CHARS: usize = 200;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum DispatchPhase {
Observe,
Decide,
}
fn honored_arm(phase: DispatchPhase, event: &str) -> Option<&'static str> {
let table = match phase {
DispatchPhase::Observe => HONORED_OUTPUT_ARM,
DispatchPhase::Decide => HONORED_DECISION_ARM,
};
table
.iter()
.find(|(name, _)| *name == event)
.and_then(|(_, arm)| *arm)
}
fn output_arm_name(output: &HookOutput) -> Option<&'static str> {
match output {
HookOutput::None => None,
HookOutput::ReplaceContext(_) => Some("replace-context"),
HookOutput::WriteManifests(_) => Some("write-manifests"),
HookOutput::Artifact(_) => Some("artifact"),
HookOutput::ReopenTask(_) => Some("reopen-task"),
HookOutput::SeedContext(_) => Some("seed-context"),
HookOutput::Deny(_) => Some("deny"),
}
}
enum OutputDisposition {
Ignore,
Honored,
Fault(&'static str),
}
fn classify_output(phase: DispatchPhase, event: &str, output: &HookOutput) -> OutputDisposition {
match output_arm_name(output) {
None => OutputDisposition::Ignore,
Some(arm) if Some(arm) == honored_arm(phase, event) => OutputDisposition::Honored,
Some(arm) => OutputDisposition::Fault(arm),
}
}
fn format_dispatch_fault(hook_name: &str, event: &str, arm: &str) -> String {
format!(
"hook '{hook_name}' returned unsupported hook-output arm '{arm}' from '{event}'; \
this event does not honor that arm, so the value was discarded"
)
}
#[derive(Debug, Clone)]
pub(crate) struct DispatchFault {
pub hook_name: String,
pub event: String,
pub arm: String,
}
struct HookInstance {
name: String,
config: HookConfig,
store: Store<HookStoreState>,
funcs: HashMap<String, wasmtime::component::Func>,
}
struct AsyncHookQueue {
name: String,
binding: HookBinding,
on_overflow: HookOverflowPolicy,
tx: mpsc::Sender<AsyncHookJob>,
worker: tokio::task::JoinHandle<()>,
dropped: u64,
}
struct AsyncHookJob {
event: HookEvent,
elapsed: Duration,
totals: HookTotals,
}
struct HookStoreState {
limits: ExecutionLimiter,
table: ResourceTable,
wasi: WasiCtx,
http: WasiHttpCtx,
http_hooks: NetworkPolicyHooks,
}
#[derive(Debug, Clone, Copy)]
struct HookTotals {
input_tokens: u64,
output_tokens: u64,
tool_calls: u32,
shell_calls: u32,
}
#[derive(Debug, Clone)]
pub(crate) struct SessionContextData {
pub capsule_name: String,
pub capsule_version: String,
pub session_id: String,
pub model: String,
pub capabilities: Vec<String>,
}
#[derive(Clone)]
pub(crate) enum HookEvent {
SessionStart,
TaskStart {
task_id: String,
context_id: String,
source: String,
input_bytes: u64,
budget_tokens: u64,
context_window: u64,
prior_tokens: u64,
},
Inference {
turn: u32,
input_tokens: u64,
output_tokens: u64,
decision: String,
tool_name: Option<String>,
prompt: Option<String>,
output: Option<String>,
tools: Option<String>,
},
ToolCall {
turn: u32,
tool_name: String,
input_bytes: u64,
input: String,
output_bytes: u64,
duration_ms: u64,
status: String,
},
Shell {
turn: u32,
binary: String,
command: String,
argv: Vec<String>,
script: Option<String>,
recipe: Option<String>,
exit_code: i32,
stdout: String,
stderr: String,
stdout_bytes: u64,
stderr_bytes: u64,
duration_ms: u64,
},
Compaction {
messages: Vec<Message>,
session_tokens: u64,
threshold: f64,
model: Option<String>,
system_prompt: Option<String>,
},
TaskEnd {
task_id: String,
exit_status: String,
},
SessionEnd {
total_turns: u32,
exit_status: String,
},
}
enum HookCallResult {
None,
Artifact(HookArtifact),
ReplaceContext(Vec<Message>),
SeedContext {
hook_name: String,
messages: Vec<Message>,
},
Reopen { hook_name: String, reason: String },
UnsupportedArm { event: String, arm: String },
}
struct DispatchOutcome {
artifacts: Vec<HookArtifact>,
replacement: Option<Vec<Message>>,
seed: Option<HookSeed>,
first_error: Option<String>,
reopen: Option<TaskReopen>,
}
#[derive(Debug, Clone)]
pub(crate) struct TaskReopen {
pub hook_name: String,
pub reason: String,
}
#[derive(Debug, Clone)]
pub(crate) struct HookSeed {
pub hook_name: String,
pub messages: Vec<Message>,
}
#[derive(Debug, Clone)]
pub(crate) struct HookArtifact {
pub hook_name: String,
pub payload: String,
}
#[derive(Debug, Clone)]
pub(crate) struct ShellDispatchInfo {
pub binary: String,
pub command: String,
pub argv: Vec<String>,
pub script: Option<String>,
pub recipe: Option<String>,
pub exit_code: i32,
pub stdout: String,
pub stderr: String,
pub stdout_bytes: u64,
pub stderr_bytes: u64,
pub duration_ms: u64,
pub resource_limit: Option<String>,
}
pub(crate) enum ResolvedCall {
Shell {
binary: String,
command: String,
argv: Vec<String>,
script: Option<String>,
recipe: Option<String>,
},
Tool {
tool_name: String,
input: String,
input_bytes: u64,
},
}
impl ResolvedCall {
pub(crate) fn event_name(&self) -> &'static str {
match self {
Self::Shell { .. } => "on-shell",
Self::Tool { .. } => "on-tool-call",
}
}
pub(crate) fn target(&self) -> &str {
match self {
Self::Shell { binary, .. } => binary,
Self::Tool { tool_name, .. } => tool_name,
}
}
fn binding(&self) -> HookBinding {
match self {
Self::Shell { .. } => HookBinding::OnShell,
Self::Tool { .. } => HookBinding::OnToolCall,
}
}
}
pub(crate) enum CallDecision {
Proceed,
Denied { hook_name: String, reason: String },
}
#[derive(Default)]
pub(crate) struct HookEnvVars<'a> {
pub otel_endpoint: Option<&'a str>,
pub eval_config_json: Option<&'a str>,
pub case_id: Option<&'a str>,
pub dataset_id: Option<&'a str>,
}
pub(crate) fn dispatch_stage(
engine: &wasmtime::Engine,
workdir: &Path,
staged_hooks: &[StagedHookArtifact],
shell_allow: Vec<String>,
env_vars: &HookEnvVars<'_>,
limits: ExecutionLimits,
) -> Result<(), RuntimeError> {
let matching: Vec<&StagedHookArtifact> = staged_hooks
.iter()
.filter(|h| {
matches!(h.config.execution_mode, HookExecutionMode::Blocking)
&& matches!(h.config.binding, HookBinding::OnStage | HookBinding::All)
})
.collect();
if matching.is_empty() {
return Ok(());
}
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| RuntimeError::Runtime(format!("failed to build on-stage runtime: {e}")))?;
rt.block_on(async {
let evt = StageEvent { shell_allow };
for staged in &matching {
match call_stage_once(engine, workdir, staged, &evt, env_vars, limits).await {
Ok(output) => match classify_output(DispatchPhase::Observe, "on-stage", &output) {
OutputDisposition::Honored => {
if let HookOutput::WriteManifests(manifests) = output {
for m in manifests {
let dir = workdir.join("tools").join(&m.binary_name);
if let Err(e) = std::fs::create_dir_all(&dir) {
log_hook_error(
workdir,
&staged.name,
&format!(
"failed to create tool dir for {}: {e}",
m.binary_name
),
)
.await;
continue;
}
let manifest_path = dir.join(PACKED_MANIFEST_ENTRY);
if let Err(e) = std::fs::write(&manifest_path, &m.content) {
log_hook_error(
workdir,
&staged.name,
&format!(
"failed to write manifest for {}: {e}",
m.binary_name
),
)
.await;
}
}
}
}
OutputDisposition::Fault(arm) => {
log_hook_error(
workdir,
&staged.name,
&format_dispatch_fault(&staged.name, "on-stage", arm),
)
.await;
}
OutputDisposition::Ignore => {}
},
Err(err) => log_hook_error(workdir, &staged.name, &err).await,
}
}
Ok(())
})
}
async fn call_stage_once(
engine: &wasmtime::Engine,
workdir: &Path,
staged: &StagedHookArtifact,
evt: &StageEvent,
env_vars: &HookEnvVars<'_>,
limits: ExecutionLimits,
) -> Result<HookOutput, String> {
let mut linker: Linker<HookStoreState> = Linker::new(engine);
wasmtime_wasi::p2::add_to_linker_async(&mut linker).map_err(|e| e.to_string())?;
wasmtime_wasi_http::p2::add_only_http_to_linker_sync(&mut linker).map_err(|e| e.to_string())?;
add_inference_to_linker(&mut linker, format!("hook:{}", staged.name), None)?;
add_task_io_to_linker(
&mut linker,
staged
.grant
.task_io_read
.then(|| Arc::new(TaskIoState::new())),
)?;
add_conversation_to_linker(
&mut linker,
staged
.grant
.conversation_read
.then(|| Arc::new(ConversationState::new(None, workdir))),
)?;
add_tokens_to_linker(&mut linker)?;
let state = HookStoreState {
limits: limits.limiter(),
table: ResourceTable::new(),
wasi: build_wasi_ctx(workdir, env_vars, &staged.grant).map_err(|e| e.to_string())?,
http: WasiHttpCtx::new(),
http_hooks: NetworkPolicyHooks {
network_allow_rules: staged.grant.network_allow_rules.clone(),
inference_gateway: None,
},
};
let mut store = Store::new(engine, state);
store.limiter(|state| &mut state.limits);
store.set_epoch_deadline(limits.deadline_ticks());
let instance = linker
.instantiate_async(&mut store, &staged.component)
.await
.map_err(|e| {
format!(
"failed to instantiate hook {}@{}: {e}",
staged.name, staged.version
)
})?;
let obs_idx = resolve_lifecycle_iface(&instance, &mut store)
.ok_or_else(|| missing_lifecycle_msg(&format!("{}@{}", staged.name, staged.version)))?;
let func = instance
.get_export_index(&mut store, Some(&obs_idx), "on-stage")
.and_then(|idx| instance.get_func(&mut store, idx))
.ok_or_else(|| format!("hook {}@{} missing on-stage", staged.name, staged.version))?;
call_stage_lift::<HookOutput>(&mut store, &func, evt, &staged.name, limits).await
}
async fn call_stage_lift<O>(
store: &mut Store<HookStoreState>,
func: &wasmtime::component::Func,
evt: &StageEvent,
hook_name: &str,
limits: ExecutionLimits,
) -> Result<O, String>
where
O: wasmtime::component::ComponentType + wasmtime::component::Lift + 'static,
{
let f = func
.typed::<(StageEvent,), (Result<O, String>,)>(&*store)
.map_err(|e| e.to_string())?;
store.set_epoch_deadline(limits.deadline_ticks());
let called = f.call_async(&mut *store, (evt.clone(),)).await;
let (result,) = match called {
Ok(result) => result,
Err(err) => {
let failure = classify_guest_failure(&err, &store.data().limits);
return Err(failure.message(&format!("hook '{hook_name}' on-stage"), &err));
}
};
result
}
impl HookRuntime {
#[allow(clippy::too_many_arguments)]
pub(crate) async fn new(
engine: &wasmtime::Engine,
workdir: &Path,
accessible_workdir: &Path,
staged_hooks: Vec<StagedHookArtifact>,
context: SessionContextData,
env_vars: HookEnvVars<'_>,
limits: ExecutionLimits,
inference: Option<Arc<HookInferenceCtx>>,
conversation_root: Option<PathBuf>,
) -> Result<Self, RuntimeError> {
let mut blocking_hooks = Vec::new();
let mut async_hooks = Vec::new();
let (fault_tx, fault_rx) = mpsc::unbounded_channel();
let host_state = HookHostState {
task_io: Arc::new(TaskIoState::new()),
conversation: Arc::new(ConversationState::new(conversation_root, workdir)),
};
for staged in staged_hooks {
let instance = instantiate_hook(
engine,
accessible_workdir,
&staged,
&env_vars,
limits,
inference.clone(),
&host_state,
)
.await?;
match staged.config.execution_mode {
HookExecutionMode::Blocking => blocking_hooks.push(instance),
HookExecutionMode::Async => {
let (tx, rx) = mpsc::channel(ASYNC_HOOK_QUEUE_DEPTH);
let worker = tokio::task::spawn_local(run_async_hook_worker(
instance,
rx,
context.clone(),
workdir.to_path_buf(),
fault_tx.clone(),
));
async_hooks.push(AsyncHookQueue {
name: staged.name.clone(),
binding: staged.config.binding.clone(),
on_overflow: staged.on_overflow,
tx,
worker,
dropped: 0,
});
}
}
}
let gates_calls = blocking_hooks.iter().any(|hook| {
hook.config.commit_policy == HookCommitPolicy::Deny
&& matches!(
hook.config.binding,
HookBinding::OnShell | HookBinding::OnToolCall
)
});
Ok(Self {
workdir: workdir.to_path_buf(),
blocking_hooks,
async_hooks,
fault_tx,
fault_rx,
context,
inference,
task_io: host_state.task_io,
conversation: host_state.conversation,
dispatch_faults: Vec::new(),
gates_calls,
started: Instant::now(),
total_input_tokens: 0,
total_output_tokens: 0,
total_tool_calls: 0,
total_shell_calls: 0,
total_turns: 0,
})
}
pub(crate) fn total_turns(&self) -> u32 {
self.total_turns
}
pub(crate) fn drain_inference_records(&self) -> Vec<HookInferenceRecord> {
self.inference
.as_ref()
.map(|ctx| ctx.drain_records())
.unwrap_or_default()
}
pub(crate) fn drain_spend_refusals(&self) -> Vec<crate::inference_import::SpendRefusalRecord> {
self.inference
.as_ref()
.map(|ctx| ctx.drain_spend_refusals())
.unwrap_or_default()
}
pub(crate) fn begin_task_attempt(&self, original: String, as_given: String) {
self.task_io.begin_task_attempt(original, as_given);
}
pub(crate) fn record_task_output(&self, text: &str) {
self.task_io.record_output(text);
}
pub(crate) fn end_task(&self) {
self.task_io.end_task();
}
pub(crate) fn drain_dispatch_faults(&mut self) -> Vec<DispatchFault> {
let mut faults = std::mem::take(&mut self.dispatch_faults);
while let Ok(fault) = self.fault_rx.try_recv() {
faults.push(fault);
}
faults
}
pub(crate) async fn drain_async_hooks(&mut self) {
self.drain_async_hooks_within(ASYNC_HOOK_DRAIN_TIMEOUT)
.await;
}
async fn drain_async_hooks_within(&mut self, budget: Duration) {
let queues = std::mem::take(&mut self.async_hooks);
if queues.is_empty() {
return;
}
let deadline = Instant::now() + budget;
for queue in queues {
let AsyncHookQueue {
name,
tx,
mut worker,
dropped,
..
} = queue;
drop(tx);
if dropped > 0 {
let summary = format!(
"hook '{name}' discarded {dropped} lifecycle event(s): its queue \
({ASYNC_HOOK_QUEUE_DEPTH} deep) was full and the capsule manifest declares \
'on_overflow: drop' for this hook"
);
log_hook_error(&self.workdir, &name, &summary).await;
}
let remaining = deadline.saturating_duration_since(Instant::now());
match tokio::time::timeout(remaining, &mut worker).await {
Ok(Ok(())) => {}
Ok(Err(join_error)) => {
let message = format!("hook '{name}' worker task failed: {join_error}");
log_hook_error(&self.workdir, &name, &message).await;
}
Err(_) => {
worker.abort();
let message = format!(
"hook '{name}' did not finish its queued lifecycle calls within the \
{}ms session-end drain budget; the remaining calls were abandoned",
budget.as_millis()
);
log_hook_error(&self.workdir, &name, &message).await;
self.dispatch_faults.push(DispatchFault {
hook_name: name,
event: FAULT_EVENT_DRAIN.to_string(),
arm: FAULT_ARM_TIMEOUT.to_string(),
});
}
}
}
}
pub(crate) async fn emit(&mut self, workdir: &Path, event: HookEvent) -> Vec<HookArtifact> {
self.dispatch(workdir, event).await.artifacts
}
pub(crate) fn gates_calls(&self) -> bool {
self.gates_calls
}
pub(crate) async fn decide(
&mut self,
workdir: &Path,
turn: u32,
call: &ResolvedCall,
) -> CallDecision {
let event_fn = call.event_name();
let binding = call.binding();
let mut decision = CallDecision::Proceed;
let mut faults: Vec<DispatchFault> = Vec::new();
for hook in &mut self.blocking_hooks {
if hook.config.commit_policy != HookCommitPolicy::Deny || hook.config.binding != binding
{
continue;
}
let name = hook.name.clone();
let called = match call {
ResolvedCall::Shell {
binary,
command,
argv,
script,
recipe,
} => {
let evt = ShellEvent {
turn,
binary: binary.clone(),
command: command.chars().take(SHELL_COMMAND_DISPLAY_CHARS).collect(),
argv: argv.clone(),
script: script.clone(),
recipe: recipe.clone(),
outcome: None,
};
call_typed(hook, event_fn, evt).await
}
ResolvedCall::Tool {
tool_name,
input,
input_bytes,
} => {
let evt = ToolEvent {
turn,
tool_name: tool_name.clone(),
input_bytes: *input_bytes,
input: input.clone(),
outcome: None,
};
call_typed(hook, event_fn, evt).await
}
};
let reason = match called {
Ok(Some(output)) => match classify_output(DispatchPhase::Decide, event_fn, &output)
{
OutputDisposition::Ignore => continue,
OutputDisposition::Honored => match output {
HookOutput::Deny(reason) if !reason.trim().is_empty() => reason,
_ => format!(
"policy hook '{name}' refused this call from '{event_fn}' but \
returned an empty reason"
),
},
OutputDisposition::Fault(arm) => {
log_hook_error(
workdir,
&name,
&format_dispatch_fault(&name, event_fn, arm),
)
.await;
faults.push(DispatchFault {
hook_name: name.clone(),
event: event_fn.to_string(),
arm: arm.to_string(),
});
format!(
"policy hook '{name}' returned hook-output arm '{arm}' from the \
'{event_fn}' decision point; only 'deny' or 'none' is answerable \
there, so the call was refused"
)
}
},
Ok(None) => format!(
"policy hook '{name}' does not export '{event_fn}'; a policy hook must \
export the function it gates, so the call was refused"
),
Err(error) => {
log_hook_error(workdir, &name, &error).await;
format!(
"policy hook '{name}' failed at the '{event_fn}' decision point: {error}"
)
}
};
decision = CallDecision::Denied {
hook_name: name,
reason,
};
break;
}
self.dispatch_faults.append(&mut faults);
decision
}
pub(crate) fn record_dispatch_fault(&mut self, fault: DispatchFault) {
self.dispatch_faults.push(fault);
}
async fn dispatch(&mut self, workdir: &Path, event: HookEvent) -> DispatchOutcome {
if let HookEvent::Inference {
input_tokens,
output_tokens,
..
} = &event
{
self.total_input_tokens = self.total_input_tokens.saturating_add(*input_tokens);
self.total_output_tokens = self.total_output_tokens.saturating_add(*output_tokens);
self.total_turns = self.total_turns.saturating_add(1);
}
if matches!(event, HookEvent::ToolCall { .. }) {
self.total_tool_calls = self.total_tool_calls.saturating_add(1);
}
if matches!(event, HookEvent::Shell { .. }) {
self.total_shell_calls = self.total_shell_calls.saturating_add(1);
}
let totals = HookTotals {
input_tokens: self.total_input_tokens,
output_tokens: self.total_output_tokens,
tool_calls: self.total_tool_calls,
shell_calls: self.total_shell_calls,
};
let mut artifacts: Vec<HookArtifact> = Vec::new();
let mut replacement: Option<Vec<Message>> = None;
let mut seed: Option<HookSeed> = None;
let mut reopen: Option<TaskReopen> = None;
let mut first_error: Option<String> = None;
let mut faults: Vec<DispatchFault> = Vec::new();
for hook in &mut self.blocking_hooks {
if !binding_matches_event(&hook.config.binding, &event) {
continue;
}
match call_hook(hook, &self.context, &event, self.started.elapsed(), totals).await {
Ok(HookCallResult::Artifact(a)) => {
artifacts.push(a);
}
Ok(HookCallResult::ReplaceContext(msgs)) => {
if replacement.is_none() {
replacement = Some(msgs);
}
}
Ok(HookCallResult::SeedContext {
hook_name,
messages,
}) => {
if seed.is_none() {
seed = Some(HookSeed {
hook_name,
messages,
});
}
}
Ok(HookCallResult::Reopen { hook_name, reason }) => {
if reopen.is_none() {
reopen = Some(TaskReopen { hook_name, reason });
}
}
Ok(HookCallResult::None) => {}
Ok(HookCallResult::UnsupportedArm { event: ev, arm }) => {
log_hook_error(
workdir,
&hook.name,
&format_dispatch_fault(&hook.name, &ev, &arm),
)
.await;
faults.push(DispatchFault {
hook_name: hook.name.clone(),
event: ev,
arm,
});
}
Err(error) => {
log_hook_error(workdir, &hook.name, &error).await;
if first_error.is_none() {
first_error = Some(format!("{}: {error}", hook.name));
}
}
}
}
self.dispatch_faults.append(&mut faults);
let elapsed = self.started.elapsed();
for queue in &mut self.async_hooks {
if !binding_matches_event(&queue.binding, &event) {
continue;
}
let job = AsyncHookJob {
event: event.clone(),
elapsed,
totals,
};
match queue.on_overflow {
HookOverflowPolicy::Drop => match queue.tx.try_send(job) {
Ok(()) => {}
Err(mpsc::error::TrySendError::Full(job)) => {
queue.dropped = queue.dropped.saturating_add(1);
let _ = self.fault_tx.send(DispatchFault {
hook_name: queue.name.clone(),
event: event_fn_name(&job.event).to_string(),
arm: FAULT_ARM_QUEUE_OVERFLOW.to_string(),
});
}
Err(mpsc::error::TrySendError::Closed(_)) => {}
},
HookOverflowPolicy::Block => {
let _ = queue.tx.send(job).await;
}
}
}
DispatchOutcome {
artifacts,
replacement,
seed,
first_error,
reopen,
}
}
pub(crate) async fn dispatch_compaction(
&mut self,
messages: Vec<Message>,
session_tokens: u64,
threshold: f64,
model: Option<String>,
system_prompt: Option<String>,
) -> Result<Option<Vec<Message>>, String> {
let workdir = self.workdir.clone();
let event = HookEvent::Compaction {
messages,
session_tokens,
threshold,
model,
system_prompt,
};
let outcome = self.dispatch(&workdir, event).await;
match outcome.first_error {
Some(error) => Err(error),
None => Ok(outcome.replacement),
}
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn dispatch_task_start(
&mut self,
task_id: String,
context_id: String,
source: String,
input_bytes: u64,
budget_tokens: u64,
context_window: u64,
prior_tokens: u64,
) -> Option<HookSeed> {
let workdir = self.workdir.clone();
self.conversation.set_context(Some(context_id.clone()));
let event = HookEvent::TaskStart {
task_id,
context_id,
source,
input_bytes,
budget_tokens,
context_window,
prior_tokens,
};
self.dispatch(&workdir, event).await.seed
}
pub(crate) async fn dispatch_task_end(
&mut self,
task_id: String,
exit_status: String,
) -> Option<TaskReopen> {
let workdir = self.workdir.clone();
let event = HookEvent::TaskEnd {
task_id,
exit_status,
};
self.dispatch(&workdir, event).await.reopen
}
}
fn binding_matches_event(binding: &HookBinding, event: &HookEvent) -> bool {
match binding {
HookBinding::All => true,
HookBinding::OnStage => false, HookBinding::OnSessionStart => matches!(event, HookEvent::SessionStart),
HookBinding::OnTaskStart => matches!(event, HookEvent::TaskStart { .. }),
HookBinding::OnInference => matches!(event, HookEvent::Inference { .. }),
HookBinding::OnToolCall => matches!(event, HookEvent::ToolCall { .. }),
HookBinding::OnShell => matches!(event, HookEvent::Shell { .. }),
HookBinding::OnCompaction => matches!(event, HookEvent::Compaction { .. }),
HookBinding::OnTaskEnd => matches!(event, HookEvent::TaskEnd { .. }),
HookBinding::OnSessionEnd => matches!(event, HookEvent::SessionEnd { .. }),
}
}
struct HookHostState {
task_io: Arc<TaskIoState>,
conversation: Arc<ConversationState>,
}
async fn instantiate_hook(
engine: &wasmtime::Engine,
project_dir: &Path,
staged: &StagedHookArtifact,
env_vars: &HookEnvVars<'_>,
limits: ExecutionLimits,
inference: Option<Arc<HookInferenceCtx>>,
host_state: &HookHostState,
) -> Result<HookInstance, RuntimeError> {
let mut linker: Linker<HookStoreState> = Linker::new(engine);
wasmtime_wasi::p2::add_to_linker_async(&mut linker)
.map_err(|err| RuntimeError::Runtime(err.to_string()))?;
wasmtime_wasi_http::p2::add_only_http_to_linker_sync(&mut linker)
.map_err(|err| RuntimeError::Runtime(err.to_string()))?;
add_inference_to_linker(&mut linker, format!("hook:{}", staged.name), inference)
.map_err(RuntimeError::Runtime)?;
add_task_io_to_linker(
&mut linker,
staged
.grant
.task_io_read
.then(|| Arc::clone(&host_state.task_io)),
)
.map_err(RuntimeError::Runtime)?;
add_conversation_to_linker(
&mut linker,
staged
.grant
.conversation_read
.then(|| Arc::clone(&host_state.conversation)),
)
.map_err(RuntimeError::Runtime)?;
add_tokens_to_linker(&mut linker).map_err(RuntimeError::Runtime)?;
let state = HookStoreState {
limits: limits.limiter(),
table: ResourceTable::new(),
wasi: build_wasi_ctx(project_dir, env_vars, &staged.grant)?,
http: WasiHttpCtx::new(),
http_hooks: NetworkPolicyHooks {
network_allow_rules: staged.grant.network_allow_rules.clone(),
inference_gateway: None,
},
};
let mut store = Store::new(engine, state);
store.limiter(|state| &mut state.limits);
store.set_epoch_deadline(limits.deadline_ticks());
let instance = linker
.instantiate_async(&mut store, &staged.component)
.await
.map_err(|err| {
RuntimeError::Runtime(format!(
"failed to instantiate hook {}@{}: {err}",
staged.name, staged.version
))
})?;
let obs_idx = resolve_lifecycle_iface(&instance, &mut store).ok_or_else(|| {
RuntimeError::Runtime(missing_lifecycle_msg(&format!(
"{}@{}",
staged.name, staged.version
)))
})?;
let funcs = resolve_hook_fns(&instance, &mut store, &obs_idx, |fn_name| {
RuntimeError::Runtime(format!(
"hook {}@{} missing function {LIFECYCLE_IFACE}#{fn_name}",
staged.name, staged.version
))
})?;
Ok(HookInstance {
funcs,
name: staged.name.clone(),
config: staged.config.clone(),
store,
})
}
fn resolve_hook_fns<E>(
instance: &wasmtime::component::Instance,
store: &mut Store<HookStoreState>,
obs_idx: &wasmtime::component::ComponentExportIndex,
missing: impl Fn(&str) -> E,
) -> Result<HashMap<String, wasmtime::component::Func>, E> {
let mut funcs = HashMap::new();
for fn_name in REQUIRED_HOOK_FNS {
let func = instance
.get_export_index(&mut *store, Some(obs_idx), fn_name)
.and_then(|idx| instance.get_func(&mut *store, idx))
.ok_or_else(|| missing(fn_name))?;
funcs.insert(fn_name.to_string(), func);
}
for fn_name in OPTIONAL_HOOK_FNS {
if let Some(func) = instance
.get_export_index(&mut *store, Some(obs_idx), fn_name)
.and_then(|idx| instance.get_func(&mut *store, idx))
{
funcs.insert(fn_name.to_string(), func);
}
}
Ok(funcs)
}
impl WasiView for HookStoreState {
fn ctx(&mut self) -> WasiCtxView<'_> {
WasiCtxView {
ctx: &mut self.wasi,
table: &mut self.table,
}
}
}
impl WasiHttpView for HookStoreState {
fn http(&mut self) -> WasiHttpCtxView<'_> {
WasiHttpCtxView {
ctx: &mut self.http,
table: &mut self.table,
hooks: &mut self.http_hooks,
}
}
}
async fn call_typed<T>(
hook: &mut HookInstance,
fn_name: &str,
arg: T,
) -> Result<Option<HookOutput>, String>
where
T: wasmtime::component::ComponentType + wasmtime::component::Lower,
{
let func = match hook.funcs.get(fn_name) {
Some(f) => *f,
None => return Ok(None),
};
let out = invoke_typed::<T, HookOutput>(hook, &func, fn_name, arg).await?;
Ok(Some(out))
}
async fn invoke_typed<T, O>(
hook: &mut HookInstance,
func: &wasmtime::component::Func,
fn_name: &str,
arg: T,
) -> Result<O, String>
where
T: wasmtime::component::ComponentType + wasmtime::component::Lower,
O: wasmtime::component::ComponentType + wasmtime::component::Lift + 'static,
{
let f = func
.typed::<(T,), (Result<O, String>,)>(&hook.store)
.map_err(|e| e.to_string())?;
let ticks = hook.store.data().limits.limits().deadline_ticks();
hook.store.set_epoch_deadline(ticks);
let called = f.call_async(&mut hook.store, (arg,)).await;
let (result,) = match called {
Ok(result) => result,
Err(err) => {
let failure = classify_guest_failure(&err, &hook.store.data().limits);
let subject = format!("hook '{}' {fn_name}", hook.name);
return Err(failure.message(&subject, &err));
}
};
result
}
async fn call_hook(
hook: &mut HookInstance,
context: &SessionContextData,
event: &HookEvent,
elapsed: Duration,
totals: HookTotals,
) -> Result<HookCallResult, String> {
let output = match event {
HookEvent::SessionStart => {
let ctx = SessionContext {
capsule_name: context.capsule_name.clone(),
capsule_version: context.capsule_version.clone(),
session_id: context.session_id.clone(),
model: context.model.clone(),
capabilities: context.capabilities.clone(),
};
call_typed(hook, "on-session-start", ctx).await?
}
HookEvent::TaskStart {
task_id,
context_id,
source,
input_bytes,
budget_tokens,
context_window,
prior_tokens,
} => {
let evt = TaskStartEvent {
task_id: task_id.clone(),
context_id: context_id.clone(),
source: source.clone(),
input_bytes: *input_bytes,
budget_tokens: *budget_tokens,
context_window: *context_window,
prior_tokens: *prior_tokens,
};
call_typed(hook, "on-task-start", evt).await?
}
HookEvent::Inference {
turn,
input_tokens,
output_tokens,
decision,
tool_name,
prompt,
output,
tools,
} => {
let evt = InferenceEvent {
turn: *turn,
input_tokens: *input_tokens,
output_tokens: *output_tokens,
decision: decision.clone(),
tool_name: tool_name.clone(),
prompt: prompt.clone(),
output: output.clone(),
tools: tools.clone(),
};
call_typed(hook, "on-inference", evt).await?
}
HookEvent::ToolCall {
turn,
tool_name,
input_bytes,
input,
output_bytes,
duration_ms,
status,
} => {
let evt = ToolEvent {
turn: *turn,
tool_name: tool_name.clone(),
input_bytes: *input_bytes,
input: input.clone(),
outcome: Some(ToolOutcome {
output_bytes: *output_bytes,
duration_ms: *duration_ms,
status: status.clone(),
}),
};
call_typed(hook, "on-tool-call", evt).await?
}
HookEvent::Shell {
turn,
binary,
command,
argv,
script,
recipe,
exit_code,
stdout,
stderr,
stdout_bytes,
stderr_bytes,
duration_ms,
} => {
let evt = ShellEvent {
turn: *turn,
binary: binary.clone(),
command: command.chars().take(SHELL_COMMAND_DISPLAY_CHARS).collect(),
argv: argv.clone(),
script: script.clone(),
recipe: recipe.clone(),
outcome: Some(ShellOutcome {
exit_code: *exit_code,
stdout: stdout.clone(),
stderr: stderr.clone(),
stdout_bytes: *stdout_bytes,
stderr_bytes: *stderr_bytes,
duration_ms: *duration_ms,
}),
};
call_typed(hook, "on-shell", evt).await?
}
HookEvent::Compaction {
messages,
session_tokens,
threshold,
model,
system_prompt,
} => {
let evt = CompactionEvent {
messages: messages.clone(),
session_tokens: *session_tokens,
threshold: *threshold,
model: model.clone(),
system_prompt: system_prompt.clone(),
};
call_typed(hook, "on-compaction", evt).await?
}
HookEvent::TaskEnd {
task_id,
exit_status,
} => {
let evt = TaskEndEvent {
task_id: task_id.clone(),
exit_status: exit_status.clone(),
};
call_typed(hook, "on-task-end", evt).await?
}
HookEvent::SessionEnd {
total_turns,
exit_status,
} => {
let evt = SessionEndEvent {
total_turns: *total_turns,
total_input_tokens: totals.input_tokens,
total_output_tokens: totals.output_tokens,
total_tool_calls: totals.tool_calls,
total_shell_calls: totals.shell_calls,
duration_ms: elapsed.as_millis().try_into().unwrap_or(u64::MAX),
exit_status: exit_status.clone(),
};
call_typed(hook, "on-session-end", evt).await?
}
};
let event_fn = event_fn_name(event);
Ok(match output {
None => HookCallResult::None,
Some(out) => match classify_output(DispatchPhase::Observe, event_fn, &out) {
OutputDisposition::Ignore => HookCallResult::None,
OutputDisposition::Honored => match out {
HookOutput::Artifact(payload) => HookCallResult::Artifact(HookArtifact {
hook_name: hook.name.clone(),
payload,
}),
HookOutput::ReplaceContext(msgs) => HookCallResult::ReplaceContext(msgs),
HookOutput::SeedContext(messages) => HookCallResult::SeedContext {
hook_name: hook.name.clone(),
messages,
},
HookOutput::ReopenTask(reason) => HookCallResult::Reopen {
hook_name: hook.name.clone(),
reason,
},
HookOutput::WriteManifests(_) | HookOutput::Deny(_) | HookOutput::None => {
HookCallResult::None
}
},
OutputDisposition::Fault(arm) => HookCallResult::UnsupportedArm {
event: event_fn.to_string(),
arm: arm.to_string(),
},
},
})
}
fn event_fn_name(event: &HookEvent) -> &'static str {
match event {
HookEvent::SessionStart => "on-session-start",
HookEvent::TaskStart { .. } => "on-task-start",
HookEvent::Inference { .. } => "on-inference",
HookEvent::ToolCall { .. } => "on-tool-call",
HookEvent::Shell { .. } => "on-shell",
HookEvent::Compaction { .. } => "on-compaction",
HookEvent::TaskEnd { .. } => "on-task-end",
HookEvent::SessionEnd { .. } => "on-session-end",
}
}
async fn run_async_hook_worker(
mut instance: HookInstance,
mut jobs: mpsc::Receiver<AsyncHookJob>,
context: SessionContextData,
workdir: PathBuf,
faults: mpsc::UnboundedSender<DispatchFault>,
) {
while let Some(job) = jobs.recv().await {
match call_hook(&mut instance, &context, &job.event, job.elapsed, job.totals).await {
Ok(HookCallResult::UnsupportedArm { event, arm }) => {
log_hook_error(
&workdir,
&instance.name,
&format_dispatch_fault(&instance.name, &event, &arm),
)
.await;
let _ = faults.send(DispatchFault {
hook_name: instance.name.clone(),
event,
arm,
});
}
Ok(_) => {}
Err(error) => {
log_hook_error(&workdir, &instance.name, &error).await;
let _ = faults.send(DispatchFault {
hook_name: instance.name.clone(),
event: event_fn_name(&job.event).to_string(),
arm: FAULT_ARM_ERROR.to_string(),
});
}
}
}
}
fn build_wasi_ctx(
root_dir: &Path,
env: &HookEnvVars<'_>,
grant: &HookCapabilityGrant,
) -> Result<WasiCtx, RuntimeError> {
let mut builder = WasiCtxBuilder::new();
builder.inherit_stdio();
if let Some(config_json) = grant.config_json.as_deref() {
builder.env(ARTIFACT_CONFIG_ENV, config_json);
}
if let Some(endpoint) = env.otel_endpoint {
builder.env("MURMUR_OTEL_ENDPOINT", endpoint);
}
if let Ok(formation_id) = std::env::var("MURMUR_FORMATION_ID") {
builder.env("MURMUR_FORMATION_ID", &formation_id);
}
if let Some(config_json) = env.eval_config_json {
builder.env("MURMUR_EVAL_CONFIG", config_json);
}
if let Some(id) = env.case_id {
builder.env("MURMUR_CASE_ID", id);
}
if let Some(id) = env.dataset_id {
builder.env("MURMUR_DATASET_ID", id);
}
if let Some(scope) = grant.filesystem_scope.as_deref() {
let scoped_dir = resolve_scoped_dir(root_dir, scope)?;
builder
.preopened_dir(&scoped_dir, ".", DirPerms::all(), FilePerms::all())
.map_err(|err| RuntimeError::wasi(scoped_dir, err.to_string()))?;
}
if let Some(state_dir) = grant.state_dir.as_deref() {
builder
.preopened_dir(
state_dir,
STATE_PREOPEN_NAME,
DirPerms::all(),
FilePerms::all(),
)
.map_err(|err| RuntimeError::wasi(state_dir.to_path_buf(), err.to_string()))?;
}
Ok(builder.build())
}
async fn log_hook_error(workdir: &Path, hook_name: &str, error: &str) {
use tokio::io::AsyncWriteExt;
let log_dir = workdir.join("logs");
let _ = tokio::fs::create_dir_all(&log_dir).await;
let path: PathBuf = log_dir.join(format!("hook-{hook_name}.log"));
let line = format!("{error}\n");
if let Ok(mut file) = tokio::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)
.await
{
let _ = file.write_all(line.as_bytes()).await;
let _ = file.flush().await;
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::inference_import::INFERENCE_IFACE_VERSIONED;
use tempfile::TempDir;
use wasmtime::component::Component;
const RETIRED_IFACE: &str = "murmur:hook/lifecycle@0.5.0";
fn hook_test_engine() -> wasmtime::Engine {
let mut config = wasmtime::Config::new();
config.wasm_component_model(true);
config.epoch_interruption(true);
wasmtime::Engine::new(&config).expect("engine builds")
}
fn hook_double(engine: &wasmtime::Engine, fn_names: &[&str]) -> Component {
hook_double_iface(engine, LIFECYCLE_IFACE, fn_names)
}
fn hook_double_iface(engine: &wasmtime::Engine, iface: &str, fn_names: &[&str]) -> Component {
let exports = fn_names
.iter()
.map(|n| format!(" (export \"{n}\" (func $f))"))
.collect::<Vec<_>>()
.join("\n");
let wat = format!(
"(component\n\
(core module $m (func (export \"f\")))\n\
(core instance $i (instantiate $m))\n\
(func $f (canon lift (core func $i \"f\")))\n\
(instance $lc\n{exports}\n)\n\
(export \"{iface}\" (instance $lc))\n\
)"
);
let bytes = wat::parse_str(&wat).expect("component WAT parses");
Component::new(engine, &bytes).expect("component double compiles")
}
fn staged_double(component: Component) -> StagedHookArtifact {
staged_double_granted(component, HookCapabilityGrant::default())
}
fn staged_double_granted(
component: Component,
grant: HookCapabilityGrant,
) -> StagedHookArtifact {
StagedHookArtifact {
name: "test-hook".to_string(),
version: "0.0.1".to_string(),
component,
config: HookConfig::default(),
grant,
on_overflow: Default::default(),
}
}
async fn new_with_hooks(
engine: &wasmtime::Engine,
workdir: &Path,
accessible: &Path,
staged: Vec<StagedHookArtifact>,
) -> Result<HookRuntime, RuntimeError> {
new_with_hooks_limited(
engine,
workdir,
accessible,
staged,
ExecutionLimits::default(),
)
.await
}
async fn new_with_hooks_limited(
engine: &wasmtime::Engine,
workdir: &Path,
accessible: &Path,
staged: Vec<StagedHookArtifact>,
limits: ExecutionLimits,
) -> Result<HookRuntime, RuntimeError> {
new_with_hooks_full(
engine,
workdir,
accessible,
staged,
limits,
HookEnvVars::default(),
)
.await
}
async fn new_with_hooks_full(
engine: &wasmtime::Engine,
workdir: &Path,
accessible: &Path,
staged: Vec<StagedHookArtifact>,
limits: ExecutionLimits,
env_vars: HookEnvVars<'_>,
) -> Result<HookRuntime, RuntimeError> {
HookRuntime::new(
engine,
workdir,
accessible,
staged,
SessionContextData {
capsule_name: "test-capsule".to_string(),
capsule_version: "0.1.0".to_string(),
session_id: "sess-test".to_string(),
model: "test-model".to_string(),
capabilities: Vec::new(),
},
env_vars,
limits,
None,
None,
)
.await
}
#[test]
fn blocking_hook_missing_task_fns_instantiates_and_skips_them() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let component = hook_double(&engine, &REQUIRED_HOOK_FNS);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double(component)],
)
.await
.expect("a hook lacking on-task-start/on-task-end must still instantiate");
assert_eq!(hooks.blocking_hooks.len(), 1);
let funcs = &hooks.blocking_hooks[0].funcs;
for name in REQUIRED_HOOK_FNS {
assert!(funcs.contains_key(name), "required fn {name} must resolve");
}
assert!(!funcs.contains_key("on-task-start"));
assert!(!funcs.contains_key("on-task-end"));
hooks
.emit(
session.path(),
HookEvent::TaskStart {
task_id: "tsk_1".to_string(),
context_id: "ctx_1".to_string(),
source: "a2a".to_string(),
input_bytes: 3,
budget_tokens: 0,
context_window: 0,
prior_tokens: 0,
},
)
.await;
hooks
.emit(
session.path(),
HookEvent::TaskEnd {
task_id: "tsk_1".to_string(),
exit_status: "ok".to_string(),
},
)
.await;
});
assert!(
!session
.path()
.join("logs")
.join("hook-test-hook.log")
.exists(),
"a hook that lacks the task exports must not log an error when task events fire"
);
}
#[test]
fn blocking_hook_with_task_fns_registers_them() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let mut names: Vec<&str> = REQUIRED_HOOK_FNS.to_vec();
names.extend_from_slice(&OPTIONAL_HOOK_FNS);
let component = hook_double(&engine, &names);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double(component)],
)
.await
.expect("component exporting task fns instantiates");
let funcs = &hooks.blocking_hooks[0].funcs;
assert!(funcs.contains_key("on-task-start"));
assert!(funcs.contains_key("on-task-end"));
});
}
fn staged_double_named(
name: &str,
binding: HookBinding,
component: Component,
) -> StagedHookArtifact {
StagedHookArtifact {
name: name.to_string(),
version: "0.0.1".to_string(),
component,
config: HookConfig {
binding,
..HookConfig::default()
},
grant: HookCapabilityGrant::default(),
on_overflow: Default::default(),
}
}
fn hook_log_lines(session: &Path, hook_name: &str) -> usize {
hook_log_text(session, hook_name).lines().count()
}
fn hook_log_text(session: &Path, hook_name: &str) -> String {
let path = session.join("logs").join(format!("hook-{hook_name}.log"));
std::fs::read_to_string(path).unwrap_or_default()
}
#[test]
fn queue_mode_fires_session_events_once_and_task_events_per_task() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let mut all_fns: Vec<&str> = REQUIRED_HOOK_FNS.to_vec();
all_fns.extend_from_slice(&OPTIONAL_HOOK_FNS);
let component = hook_double(&engine, &all_fns);
let staged = vec![
staged_double_named("sess-start", HookBinding::OnSessionStart, component.clone()),
staged_double_named("task-start", HookBinding::OnTaskStart, component.clone()),
staged_double_named("task-end", HookBinding::OnTaskEnd, component.clone()),
staged_double_named("sess-end", HookBinding::OnSessionEnd, component.clone()),
];
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = new_with_hooks(&engine, session.path(), accessible.path(), staged)
.await
.expect("all doubles export the six required fns and instantiate");
hooks.emit(session.path(), HookEvent::SessionStart).await;
for t in 0..3u32 {
hooks
.emit(
session.path(),
HookEvent::TaskStart {
task_id: format!("tsk_{t}"),
context_id: format!("ctx_{t}"),
source: "a2a".to_string(),
input_bytes: 1,
budget_tokens: 0,
context_window: 0,
prior_tokens: 0,
},
)
.await;
hooks
.emit(
session.path(),
HookEvent::Inference {
turn: 0,
input_tokens: 0,
output_tokens: 0,
decision: "end_turn".to_string(),
tool_name: None,
prompt: None,
output: None,
tools: None,
},
)
.await;
hooks
.emit(
session.path(),
HookEvent::TaskEnd {
task_id: format!("tsk_{t}"),
exit_status: "ok".to_string(),
},
)
.await;
}
let total_turns = hooks.total_turns();
assert_eq!(
total_turns, 3,
"one turn accumulated per inference across tasks"
);
hooks
.emit(
session.path(),
HookEvent::SessionEnd {
total_turns,
exit_status: "ok".to_string(),
},
)
.await;
});
assert_eq!(
hook_log_lines(session.path(), "sess-start"),
1,
"on-session-start fires once"
);
assert_eq!(
hook_log_lines(session.path(), "sess-end"),
1,
"on-session-end fires once"
);
assert_eq!(
hook_log_lines(session.path(), "task-start"),
3,
"on-task-start fires per task"
);
assert_eq!(
hook_log_lines(session.path(), "task-end"),
3,
"on-task-end fires per task"
);
}
#[test]
fn blocking_hook_missing_required_fn_fails_new() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let names: Vec<&str> = REQUIRED_HOOK_FNS
.iter()
.copied()
.filter(|n| *n != "on-session-start")
.collect();
let component = hook_double(&engine, &names);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let msg = match new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double(component)],
)
.await
{
Ok(_) => panic!("missing a required lifecycle fn must fail instantiation"),
Err(e) => e.to_string(),
};
assert!(
msg.contains("on-session-start"),
"error must name the missing function, got: {msg}"
);
});
}
#[test]
fn versioned_hook_double_instantiates_and_registers_fns() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let mut names: Vec<&str> = REQUIRED_HOOK_FNS.to_vec();
names.extend_from_slice(&OPTIONAL_HOOK_FNS);
let component = hook_double_iface(&engine, LIFECYCLE_IFACE, &names);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double(component)],
)
.await
.expect("a hook exporting the versioned lifecycle name must instantiate");
assert_eq!(hooks.blocking_hooks.len(), 1);
let funcs = &hooks.blocking_hooks[0].funcs;
for name in REQUIRED_HOOK_FNS {
assert!(funcs.contains_key(name), "required fn {name} must resolve");
}
assert!(funcs.contains_key("on-task-start"));
assert!(funcs.contains_key("on-task-end"));
});
}
#[test]
fn unversioned_only_hook_double_fails_hard() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let component = hook_double_iface(&engine, "murmur:hook/lifecycle", &REQUIRED_HOOK_FNS);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let msg = match new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double(component)],
)
.await
{
Ok(_) => panic!("a hook exporting only the unversioned lifecycle name must fail"),
Err(e) => e.to_string(),
};
assert!(
msg.contains(LIFECYCLE_IFACE),
"error must name the versioned lifecycle export, got: {msg}"
);
assert!(
msg.contains("rebuild"),
"error must hint at rebuilding the hook, got: {msg}"
);
});
}
#[test]
fn hook_missing_lifecycle_export_fails() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let component =
hook_double_iface(&engine, "murmur:hook/lifecycle@9.9.9", &REQUIRED_HOOK_FNS);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let msg = match new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double(component)],
)
.await
{
Ok(_) => panic!("a component exporting neither probed lifecycle name must fail"),
Err(e) => e.to_string(),
};
assert!(
msg.contains(LIFECYCLE_IFACE),
"error must name the missing lifecycle export, got: {msg}"
);
});
}
fn hook_spin_double(engine: &wasmtime::Engine) -> Component {
hook_spin_double_iface(engine, LIFECYCLE_IFACE)
}
fn hook_spin_double_iface(engine: &wasmtime::Engine, iface: &str) -> Component {
let exports = REQUIRED_HOOK_FNS
.iter()
.map(|n| format!(" (export \"{n}\" (func $f))"))
.collect::<Vec<_>>()
.join("\n");
let wat = format!(
r#"(component
(core module $m
(memory (export "memory") 1)
(func (export "realloc") (param i32 i32 i32 i32) (result i32) i32.const 8)
(func (export "spin")
(param i32 i32 i32 i32 i32 i32 i32 i32 i32 i32) (result i32)
(loop $l (br $l))
unreachable)
)
(core instance $i (instantiate $m))
(alias core export $i "memory" (core memory $mem))
(alias core export $i "realloc" (core func $realloc))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $session-context (record
(field "capsule-name" string)
(field "capsule-version" string)
(field "session-id" string)
(field "model" string)
(field "capabilities" (list string))))
(type $ft (func (param "ctx" $session-context) (result (result $hook-output (error string)))))
(func $f (type $ft)
(canon lift (core func $i "spin")
(memory $mem) (realloc $realloc) string-encoding=utf8))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "session-context" (type $session-context))
{exports}
)
(export "{iface}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("spin component WAT parses");
Component::new(engine, &bytes).expect("spin component double compiles")
}
fn compaction_component_from_core(engine: &wasmtime::Engine, core_body: &str) -> Component {
let stubs = REQUIRED_HOOK_FNS
.iter()
.filter(|n| **n != "on-compaction")
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let iface = LIFECYCLE_IFACE;
let wat = format!(
r#"(component
(core module $m
{core_body}
)
(core instance $i (instantiate $m))
(alias core export $i "memory" (core memory $mem))
(alias core export $i "realloc" (core func $realloc))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $compaction-event (record
(field "messages" (list $message))
(field "session-tokens" u64)
(field "threshold" f64)
(field "model" (option string))
(field "system-prompt" (option string))))
(type $ft (func (param "event" $compaction-event) (result (result $hook-output (error string)))))
(func $oc (type $ft)
(canon lift (core func $i "oncompact") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "compaction-event" (type $compaction-event))
(export "on-compaction" (func $oc))
{stubs}
)
(export "{iface}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("compaction component WAT parses");
Component::new(engine, &bytes).expect("compaction component double compiles")
}
const MODEL_ABSENT_SENTINEL: &str = "<<absent>>";
fn hook_compaction_echo_model_double(engine: &wasmtime::Engine) -> Component {
hook_compaction_echo_double(engine, "model", "model")
}
fn hook_compaction_echo_system_prompt_double(engine: &wasmtime::Engine) -> Component {
hook_compaction_echo_double(engine, "sp", "system-prompt")
}
fn hook_compaction_echo_double(
engine: &wasmtime::Engine,
field: &str,
role: &str,
) -> Component {
let sentinel_len = MODEL_ABSENT_SENTINEL.len();
let role_len = role.len();
let core = format!(
r#" (memory (export "memory") 1)
(global $bump (mut i32) (i32.const 1024))
(data (i32.const 240) "{role}")
(data (i32.const 264) "{MODEL_ABSENT_SENTINEL}")
(func (export "realloc") (param i32 i32 i32 i32) (result i32)
(local $r i32)
(global.set $bump
(i32.and (i32.add (global.get $bump) (i32.const 7)) (i32.const -8)))
(local.set $r (global.get $bump))
(global.set $bump (i32.add (global.get $bump) (local.get 3)))
(local.get $r))
(func (export "oncompact")
(param $msgs-ptr i32) (param $msgs-len i32)
(param $tokens i64) (param $threshold f64)
(param $model-some i32) (param $model-ptr i32) (param $model-len i32)
(param $sp-some i32) (param $sp-ptr i32) (param $sp-len i32)
(result i32)
(i32.store (i32.const 128) (i32.const 0))
(i32.store (i32.const 132) (i32.const 1))
(i32.store (i32.const 136) (i32.const 160))
(i32.store (i32.const 140) (i32.const 1))
(i32.store (i32.const 160) (i32.const 240))
(i32.store (i32.const 164) (i32.const {role_len}))
(i32.store (i32.const 168)
(select (local.get ${field}-ptr) (i32.const 264) (local.get ${field}-some)))
(i32.store (i32.const 172)
(select (local.get ${field}-len) (i32.const {sentinel_len}) (local.get ${field}-some)))
(i32.store (i32.const 176) (i32.const 0))
(i32.store (i32.const 188) (i32.const 0))
(i32.const 128))
(func (export "noop"))"#
);
compaction_component_from_core(engine, &core)
}
fn hook_compaction_err_double(engine: &wasmtime::Engine) -> Component {
let core = r#" (memory (export "memory") 1)
(data (i32.const 300) "boom")
(func (export "realloc") (param i32 i32 i32 i32) (result i32) i32.const 512)
(func (export "oncompact")
(param i32 i32 i64 f64 i32 i32 i32 i32 i32 i32) (result i32)
(i32.store (i32.const 128) (i32.const 1))
(i32.store (i32.const 132) (i32.const 300))
(i32.store (i32.const 136) (i32.const 4))
(i32.const 128))
(func (export "noop"))"#;
compaction_component_from_core(engine, core)
}
async fn dispatch_compaction_with(
session: &Path,
accessible: &Path,
engine: &wasmtime::Engine,
double: Component,
model: Option<String>,
system_prompt: Option<String>,
) -> Result<Option<Vec<Message>>, String> {
let mut hooks = new_with_hooks(
engine,
session,
accessible,
vec![staged_double_named(
"compactor",
HookBinding::OnCompaction,
double,
)],
)
.await
.expect("double instantiates");
hooks
.dispatch_compaction(
vec![Message {
role: "user".to_string(),
content: "hello".to_string(),
id: None,
source_id: None,
}],
1234,
0.98,
model,
system_prompt,
)
.await
}
#[test]
fn compaction_hook_receives_manifest_model() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(dispatch_compaction_with(
session.path(),
accessible.path(),
&engine,
hook_compaction_echo_model_double(&engine),
Some("claude-haiku-4-5".to_string()),
None,
));
let messages = result.expect("hook succeeded").expect("replace-context");
assert_eq!(messages[0].role, "model");
assert_eq!(messages[0].content, "claude-haiku-4-5");
}
#[test]
fn compaction_hook_receives_manifest_system_prompt() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let prompt = "task = X, currently editing Y, already tried Z.";
let result = rt.block_on(dispatch_compaction_with(
session.path(),
accessible.path(),
&engine,
hook_compaction_echo_system_prompt_double(&engine),
None,
Some(prompt.to_string()),
));
let messages = result.expect("hook succeeded").expect("replace-context");
assert_eq!(messages[0].role, "system-prompt");
assert_eq!(messages[0].content, prompt);
}
#[test]
fn compaction_hook_receives_none_when_system_prompt_unset() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(dispatch_compaction_with(
session.path(),
accessible.path(),
&engine,
hook_compaction_echo_system_prompt_double(&engine),
Some("claude-haiku-4-5".to_string()),
None,
));
let messages = result.expect("hook succeeded").expect("replace-context");
assert_eq!(messages[0].content, MODEL_ABSENT_SENTINEL);
}
#[test]
fn compaction_hook_receives_none_when_model_unset() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(dispatch_compaction_with(
session.path(),
accessible.path(),
&engine,
hook_compaction_echo_model_double(&engine),
None,
None,
));
let messages = result.expect("hook succeeded").expect("replace-context");
assert_eq!(messages[0].content, MODEL_ABSENT_SENTINEL);
}
#[test]
fn compaction_hook_error_surfaces_to_caller() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(dispatch_compaction_with(
session.path(),
accessible.path(),
&engine,
hook_compaction_err_double(&engine),
None,
None,
));
let error = result.expect_err("a failing compaction hook must not read as Ok(None)");
assert!(
error.contains("compactor") && error.contains("boom"),
"error must name the hook and carry its message, got {error:?}"
);
assert_eq!(
hook_log_lines(session.path(), "compactor"),
1,
"per-hook error logging is unchanged"
);
}
#[test]
fn compaction_with_no_bound_hook_is_ok_none() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let result = rt.block_on(async {
let mut hooks = new_with_hooks(&engine, session.path(), accessible.path(), vec![])
.await
.expect("empty hook set");
hooks
.dispatch_compaction(Vec::new(), 1234, 0.98, None, None)
.await
});
assert!(matches!(result, Ok(None)));
}
async fn dispatch_compaction_against(
session: &Path,
accessible: &Path,
engine: &wasmtime::Engine,
) {
let core = r#" (memory (export "memory") 1)
(func (export "realloc") (param i32 i32 i32 i32) (result i32) i32.const 512)
(func (export "oncompact") (param i32 i32 i64 f64 i32 i32 i32 i32 i32 i32) (result i32)
(i32.store (i32.const 128) (i32.const 0))
(i32.store (i32.const 132) (i32.const 0))
(i32.const 128))
(func (export "noop"))"#;
let mut hooks = new_with_hooks(
engine,
session,
accessible,
vec![staged_double_named(
"compactor",
HookBinding::OnCompaction,
compaction_component_from_core(engine, core),
)],
)
.await
.expect("a current-version hook must instantiate");
let replacement = hooks
.dispatch_compaction(
vec![Message {
role: "user".to_string(),
content: "hello".to_string(),
id: None,
source_id: None,
}],
1234,
0.98,
None,
None,
)
.await;
assert!(
matches!(replacement, Ok(None)),
"the double returns hook-output::none, so no context replacement and no error"
);
}
#[test]
fn current_hook_receives_five_field_compaction_event() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(dispatch_compaction_against(
session.path(),
accessible.path(),
&engine,
));
assert_eq!(
hook_log_lines(session.path(), "compactor"),
0,
"on-compaction must be received cleanly, not as an ABI mismatch"
);
}
#[test]
fn retired_version_hook_fails_to_instantiate() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let msg = match new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"legacy",
HookBinding::All,
hook_spin_double_iface(&engine, RETIRED_IFACE),
)],
)
.await
{
Ok(_) => panic!("a hook exporting a retired lifecycle version must not resolve"),
Err(e) => e.to_string(),
};
assert!(
msg.contains(LIFECYCLE_IFACE),
"error must name the accepted lifecycle export, got: {msg}"
);
assert!(
msg.contains("rebuild"),
"error must hint at rebuilding the hook, got: {msg}"
);
});
}
fn hook_shell_double(engine: &wasmtime::Engine, expect_binary: &str) -> Component {
assert!(
expect_binary.len() < 128,
"expected-binary data segment would overlap the return area at 128"
);
let expect_len = expect_binary.len();
let stubs = REQUIRED_HOOK_FNS
.iter()
.filter(|n| **n != "on-shell")
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let iface = LIFECYCLE_IFACE;
let wat = format!(
r#"(component
(core module $m
(memory (export "memory") 1)
(data (i32.const 0) "{expect_binary}")
(global $next (mut i32) (i32.const 1024))
(func (export "realloc") (param i32 i32 i32) (param $size i32) (result i32)
(local $at i32)
(local.set $at (global.get $next))
(global.set $next (i32.add (local.get $at) (local.get $size)))
(local.get $at))
(func (export "onshell") (param $event i32) (result i32)
(local $binp i32)
(local $i i32)
(local.set $binp (i32.load (i32.add (local.get $event) (i32.const 4))))
(if (i32.ne
(i32.load (i32.add (local.get $event) (i32.const 8)))
(i32.const {expect_len}))
(then unreachable))
(block $done
(loop $l
(br_if $done (i32.ge_u (local.get $i) (i32.const {expect_len})))
(if (i32.ne
(i32.load8_u (i32.add (local.get $binp) (local.get $i)))
(i32.load8_u (local.get $i)))
(then unreachable))
(local.set $i (i32.add (local.get $i) (i32.const 1)))
(br $l)))
(i32.store (i32.const 128) (i32.const 0))
(i32.store (i32.const 132) (i32.const 0))
(i32.const 128))
(func (export "noop"))
)
(core instance $i (instantiate $m))
(alias core export $i "memory" (core memory $mem))
(alias core export $i "realloc" (core func $realloc))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $shell-outcome (record
(field "exit-code" s32)
(field "stdout" string)
(field "stderr" string)
(field "stdout-bytes" u64)
(field "stderr-bytes" u64)
(field "duration-ms" u64)))
(type $shell-event (record
(field "turn" u32)
(field "binary" string)
(field "command" string)
(field "argv" (list string))
(field "script" (option string))
(field "recipe" (option string))
(field "outcome" (option $shell-outcome))))
(type $ft (func (param "event" $shell-event) (result (result $hook-output (error string)))))
(func $sh (type $ft)
(canon lift (core func $i "onshell") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "shell-outcome" (type $shell-outcome))
(export "shell-event" (type $shell-event))
(export "on-shell" (func $sh))
{stubs}
)
(export "{iface}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("shell component WAT parses");
Component::new(engine, &bytes).expect("shell component double compiles")
}
async fn dispatch_shell_against(
session: &Path,
accessible: &Path,
engine: &wasmtime::Engine,
component: Component,
binary: &str,
) {
let mut hooks = new_with_hooks(
engine,
session,
accessible,
vec![staged_double_named(
"sheller",
HookBinding::OnShell,
component,
)],
)
.await
.expect("a current-version hook must instantiate");
hooks
.emit(
session,
HookEvent::Shell {
turn: 1,
binary: binary.to_string(),
command: "-q tests/".to_string(),
argv: vec!["-q".to_string(), "tests/".to_string()],
script: None,
recipe: None,
exit_code: 0,
stdout: "ok".to_string(),
stderr: String::new(),
stdout_bytes: 2,
stderr_bytes: 0,
duration_ms: 12,
},
)
.await;
}
#[test]
fn current_hook_receives_the_invoked_binary_in_shell_event() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(dispatch_shell_against(
session.path(),
accessible.path(),
&engine,
hook_shell_double(&engine, "/usr/bin/pytest"),
"/usr/bin/pytest",
));
assert_eq!(
hook_log_lines(session.path(), "sheller"),
0,
"a current-version hook must receive on-shell cleanly with the exact binary the host sent"
);
}
#[test]
fn shell_double_rejects_a_binary_it_did_not_expect() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(dispatch_shell_against(
session.path(),
accessible.path(),
&engine,
hook_shell_double(&engine, "/usr/bin/pytest"),
"/usr/bin/cargo",
));
let log = std::fs::read_to_string(session.path().join("logs").join("hook-sheller.log"))
.expect("the mismatch must be logged");
assert!(
log.contains("on-shell trapped"),
"the double must trap *inside the guest* on an unexpected binary — a `.typed()` \
mismatch (which reports differently) would prove nothing about the field's \
value; got: {log}"
);
}
#[test]
fn hook_importing_run_inference_receives_the_driver_completion() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let driver = crate::inference_import::test_support::driver_double(
&engine,
0,
r#"{"stop_reason":"end_turn","content":[{"type":"text","text":"hello from driver"}]}"#,
);
let ctx = Arc::new(HookInferenceCtx {
driver_name: "mock-driver".to_string(),
driver_component: driver,
model: "manifest-model".to_string(),
engine: engine.clone(),
accessible_workdir: accessible.path().to_path_buf(),
workdir: accessible.path().to_path_buf(),
inference_env: Vec::new(),
capability_policy: crate::types::CapabilityPolicy::default(),
network_allow_rules: Vec::new(),
driver_grant: None,
inference_gateway: None,
spend: Arc::new(crate::spend::SpendMeter::unlimited()),
records: std::sync::Mutex::new(Vec::new()),
spend_refusals: std::sync::Mutex::new(Vec::new()),
});
let rt = tokio::runtime::Runtime::new().unwrap();
let artifacts = rt.block_on(async {
let mut hooks = HookRuntime::new(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"caller",
HookBinding::OnInference,
hook_inference_caller_double(&engine),
)],
SessionContextData {
capsule_name: "test-capsule".to_string(),
capsule_version: "0.1.0".to_string(),
session_id: "sess-test".to_string(),
model: "manifest-model".to_string(),
capabilities: Vec::new(),
},
HookEnvVars::default(),
ExecutionLimits::default(),
Some(Arc::clone(&ctx)),
None,
)
.await
.expect("a hook importing murmur:runtime/inference must link");
hooks.emit(session.path(), inference_event()).await
});
assert_eq!(artifacts.len(), 1, "the hook returns run-inference's text");
assert_eq!(artifacts[0].payload, "hello from driver");
let records = ctx.drain_records();
assert_eq!(records.len(), 1);
assert_eq!(records[0].origin.source, "hook:caller");
assert_eq!(records[0].origin.model, "manifest-model");
}
#[test]
fn hook_run_inference_without_driver_gets_clear_error() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let artifacts = rt.block_on(async {
let mut hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"caller",
HookBinding::OnInference,
hook_inference_caller_double(&engine),
)],
)
.await
.expect("the import is defined even with no driver, so the hook links");
hooks.emit(session.path(), inference_event()).await
});
assert_eq!(artifacts.len(), 1);
assert!(
artifacts[0]
.payload
.contains("inference driver is not configured")
&& artifacts[0].payload.contains("inference.driver.artifact"),
"got: {}",
artifacts[0].payload
);
}
fn inference_event() -> HookEvent {
HookEvent::Inference {
turn: 0,
input_tokens: 0,
output_tokens: 0,
decision: "end_turn".to_string(),
tool_name: None,
prompt: None,
output: None,
tools: None,
}
}
fn hook_inference_caller_double(engine: &wasmtime::Engine) -> Component {
let stubs = REQUIRED_HOOK_FNS
.iter()
.filter(|n| **n != "on-inference")
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let wat = format!(
r#"(component
(import "{INFERENCE_IFACE_VERSIONED}" (instance $inf
(type (option string))
(type (record
(field "role" string)
(field "content" string)
(field "id" 0)
(field "source-id" 0)))
(export "message" (type (eq 1)))
(type (list 2))
(type (record
(field "messages" 3)
(field "system-prompt" 0)
(field "model" 0)))
(export "inference-request" (type (eq 4)))
(type (record
(field "text" string)
(field "model-used" string)
(field "input-tokens" u64)
(field "output-tokens" u64)))
(export "inference-response" (type (eq 6)))
(type (result 7 (error string)))
(export "run-inference" (func (param "request" 5) (result 8)))
))
(alias export $inf "run-inference" (func $runi))
(core module $libc
(memory (export "memory") 1)
(global $bump (mut i32) (i32.const 1024))
(func (export "realloc") (param i32 i32 i32 i32) (result i32)
(local $p i32)
(local.set $p (i32.and (i32.add (global.get $bump) (i32.const 7)) (i32.const -8)))
(global.set $bump (i32.add (local.get $p) (i32.add (local.get 3) (i32.const 8))))
(local.get $p))
)
(core instance $li (instantiate $libc))
(alias core export $li "memory" (core memory $mem))
(alias core export $li "realloc" (core func $realloc))
(core func $run_lowered
(canon lower (func $runi) (memory $mem) (realloc $realloc) string-encoding=utf8))
(core module $m
(import "libc" "memory" (memory 1))
(import "inf" "run" (func $run (param i32 i32 i32 i32 i32 i32 i32 i32 i32)))
(func (export "oninf") (param i32) (result i32)
(call $run
(i32.const 0) (i32.const 0)
(i32.const 0) (i32.const 0) (i32.const 0)
(i32.const 0) (i32.const 0) (i32.const 0)
(i32.const 256))
(i32.store (i32.const 128) (i32.const 0))
(i32.store (i32.const 132) (i32.const 3))
(i32.store (i32.const 136) (i32.load (i32.const 264)))
(i32.store (i32.const 140) (i32.load (i32.const 268)))
(i32.const 128))
(func (export "noop"))
)
(core instance $i (instantiate $m
(with "libc" (instance $li))
(with "inf" (instance (export "run" (func $run_lowered))))))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $inference-event (record
(field "turn" u32)
(field "input-tokens" u64)
(field "output-tokens" u64)
(field "decision" string)
(field "tool-name" (option string))
(field "prompt" (option string))
(field "output" (option string))
(field "tools" (option string))))
(type $ft (func (param "event" $inference-event) (result (result $hook-output (error string)))))
(func $oninf (type $ft)
(canon lift (core func $i "oninf") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "inference-event" (type $inference-event))
(export "on-inference" (func $oninf))
{stubs}
)
(export "{LIFECYCLE_IFACE}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("inference-caller component WAT parses");
Component::new(engine, &bytes).expect("inference-caller component double compiles")
}
#[test]
fn blocking_hook_that_spins_is_interrupted_and_does_not_abort_dispatch() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let _ticker = crate::limits::EpochTicker::spawn(&engine);
let limits = ExecutionLimits {
deadline_seconds: 1,
..ExecutionLimits::default()
};
let staged = vec![
staged_double_named("spinner", HookBinding::All, hook_spin_double(&engine)),
staged_double_named(
"bystander",
HookBinding::All,
hook_double(&engine, &REQUIRED_HOOK_FNS),
),
];
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks =
new_with_hooks_limited(&engine, session.path(), accessible.path(), staged, limits)
.await
.expect("both doubles export every required fn, so both instantiate");
hooks.emit(session.path(), HookEvent::SessionStart).await;
});
let spinner_log =
std::fs::read_to_string(session.path().join("logs").join("hook-spinner.log"))
.expect("the interrupted hook must have logged via log_hook_error");
assert!(
spinner_log.contains("exceeded its 1s execution deadline"),
"the log must name the deadline that fired rather than read as a generic \
trap; got: {spinner_log}"
);
assert!(
session
.path()
.join("logs")
.join("hook-bystander.log")
.exists(),
"dispatch must continue to the next hook after a deadline trap"
);
}
use crate::tokens_import::TOKENS_IFACE_VERSIONED;
const TOKEN_PROBE: &str =
"the quick brown fox jumps over the lazy dog, and then counts its own tokens";
fn hook_token_counter_double(engine: &wasmtime::Engine) -> Component {
let stubs = REQUIRED_HOOK_FNS
.iter()
.filter(|n| **n != "on-session-start")
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let probe_len = TOKEN_PROBE.len();
let wat = format!(
r#"(component
(import "{TOKENS_IFACE_VERSIONED}" (instance $tk
(export "count" (func (param "text" string) (result u64)))
))
(alias export $tk "count" (func $count))
(core module $libc
(memory (export "memory") 1)
(global $bump (mut i32) (i32.const 1024))
(func (export "realloc") (param i32 i32 i32 i32) (result i32)
(local $p i32)
(local.set $p (i32.and (i32.add (global.get $bump) (i32.const 7)) (i32.const -8)))
(global.set $bump (i32.add (local.get $p) (i32.add (local.get 3) (i32.const 8))))
(local.get $p))
)
(core instance $li (instantiate $libc))
(alias core export $li "memory" (core memory $mem))
(alias core export $li "realloc" (core func $realloc))
(core func $count_lowered
(canon lower (func $count) (memory $mem) (realloc $realloc) string-encoding=utf8))
(core module $m
(import "libc" "memory" (memory 1))
(import "tok" "count" (func $count (param i32 i32) (result i64)))
(data (i32.const 300) "{TOKEN_PROBE}")
(global $cur (mut i32) (i32.const 400))
(func $put (param $b i32)
(i32.store8 (global.get $cur) (local.get $b))
(global.set $cur (i32.add (global.get $cur) (i32.const 1))))
(func $dec (param $v i32)
(if (i32.ge_u (local.get $v) (i32.const 10))
(then (call $dec (i32.div_u (local.get $v) (i32.const 10)))))
(call $put (i32.add (i32.const 48) (i32.rem_u (local.get $v) (i32.const 10)))))
(func (export "onstart") (param i32 i32 i32 i32 i32 i32 i32 i32 i32 i32) (result i32)
(call $dec
(i32.wrap_i64 (call $count (i32.const 300) (i32.const {probe_len}))))
(i32.store (i32.const 128) (i32.const 1))
(i32.store (i32.const 132) (i32.const 400))
(i32.store (i32.const 136) (i32.sub (global.get $cur) (i32.const 400)))
(i32.const 128))
(func (export "noop"))
)
(core instance $i (instantiate $m
(with "libc" (instance $li))
(with "tok" (instance (export "count" (func $count_lowered))))))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $session-context (record
(field "capsule-name" string)
(field "capsule-version" string)
(field "session-id" string)
(field "model" string)
(field "capabilities" (list string))))
(type $fs (func (param "ctx" $session-context) (result (result $hook-output (error string)))))
(func $os (type $fs)
(canon lift (core func $i "onstart") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "session-context" (type $session-context))
(export "on-session-start" (func $os))
{stubs}
)
(export "{LIFECYCLE_IFACE}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("token-counter component WAT parses");
Component::new(engine, &bytes).expect("token-counter component double compiles")
}
#[test]
fn tokens_count_answers_a_hook_with_the_hosts_own_number() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"counter",
HookBinding::OnSessionStart,
hook_token_counter_double(&engine),
)],
)
.await
.expect("a tokens-importing hook instantiates like any other");
hooks.emit(session.path(), HookEvent::SessionStart).await;
});
let expected = crate::agent::count_tokens(TOKEN_PROBE);
assert!(
expected > 0,
"the probe must actually tokenize to something"
);
assert!(
hook_log_text(session.path(), "counter").contains(&expected.to_string()),
"the guest must report the host's own count ({expected}), got {:?}",
hook_log_text(session.path(), "counter")
);
}
#[test]
fn tokens_count_is_registered_on_the_staging_linker_too() {
let session = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_double_named(
"counter",
HookBinding::OnStage,
hook_token_counter_double(&engine),
);
dispatch_stage(
&engine,
session.path(),
std::slice::from_ref(&staged),
Vec::new(),
&HookEnvVars::default(),
ExecutionLimits::default(),
)
.expect("on-stage dispatch itself never fails on a hook-returned error");
let log = hook_log_text(session.path(), "counter");
assert_eq!(
log.trim_end(),
"hook counter@0.0.1 missing on-stage",
"the hook must have instantiated and been dispatched, not failed to link"
);
}
use crate::task_io_import::test_support::{probe_double, reader_double, REPORT_SEP};
fn staged_task_io_hook(
name: &str,
binding: HookBinding,
component: Component,
task_io_read: bool,
) -> StagedHookArtifact {
StagedHookArtifact {
grant: HookCapabilityGrant {
network_allow_rules: Vec::new(),
filesystem_scope: None,
task_io_read,
conversation_read: false,
state_store: None,
state_dir: None,
config_json: None,
},
..staged_double_named(name, binding, component)
}
}
fn probe_lines(session: &Path, hook_name: &str) -> Vec<String> {
let path = session.join("logs").join(format!("hook-{hook_name}.log"));
std::fs::read_to_string(path)
.map(|s| s.lines().map(str::to_string).collect())
.unwrap_or_default()
}
#[test]
fn the_scope_table_holds_at_every_lifecycle_event() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_task_io_hook("probe", HookBinding::All, probe_double(&engine), true);
let mut on_stage = staged.clone();
on_stage.config.binding = HookBinding::OnStage;
dispatch_stage(
&engine,
session.path(),
std::slice::from_ref(&on_stage),
Vec::new(),
&HookEnvVars::default(),
ExecutionLimits::default(),
)
.expect("on-stage dispatch itself never fails on a hook-returned error");
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("a task-io-importing hook instantiates like any other");
hooks.emit(session.path(), HookEvent::SessionStart).await;
hooks
.emit(
session.path(),
HookEvent::TaskStart {
task_id: "tsk_1".to_string(),
context_id: "ctx_1".to_string(),
source: "task_md".to_string(),
input_bytes: 4,
budget_tokens: 0,
context_window: 0,
prior_tokens: 0,
},
)
.await;
hooks.begin_task_attempt("task text".to_string(), "task text".to_string());
hooks.emit(session.path(), inference_event()).await;
hooks.record_task_output("the result");
hooks
.dispatch_task_end("tsk_1".to_string(), "ok".to_string())
.await;
hooks.end_task();
hooks
.emit(
session.path(),
HookEvent::SessionEnd {
total_turns: 1,
exit_status: "ok".to_string(),
},
)
.await;
});
assert_eq!(
probe_lines(session.path(), "probe"),
vec![
"on-stage in=!1,out=!1",
"on-session-start in=!1,out=!1",
"on-task-start in=!1,out=!1",
"on-inference in=9,out=!2",
"on-task-end in=9,out=10",
"on-session-end in=!1,out=!1",
],
"this list is the table in wit/hook/deps/murmur-task-io/read.wit — an edit to \
either must move both"
);
}
#[test]
fn an_ungranted_hook_links_and_reads_not_granted() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_task_io_hook("probe", HookBinding::All, probe_double(&engine), false);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect(
"an ungranted importer still links — the import is defined, not backed",
);
hooks.begin_task_attempt("task text".to_string(), "task text".to_string());
hooks.record_task_output("the result");
hooks
.dispatch_task_end("tsk_1".to_string(), "ok".to_string())
.await;
});
assert_eq!(
probe_lines(session.path(), "probe"),
vec!["on-task-end in=!0,out=!0"],
"a denied read is a value the hook branches on, and it never leaks the length"
);
}
#[test]
fn a_granted_reader_returns_what_it_read_as_reopen_task() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged =
staged_task_io_hook("gate", HookBinding::OnTaskEnd, reader_double(&engine), true);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("reader double instantiates");
hooks.begin_task_attempt("pristine".to_string(), "pristine + feedback".to_string());
hooks.record_task_output("agent said this");
let reopen = hooks
.dispatch_task_end("tsk_1".to_string(), "ok".to_string())
.await
.expect("the reader double always returns reopen-task");
assert_eq!(
reopen.reason.split(REPORT_SEP).collect::<Vec<_>>(),
[
"A=pristine + feedback",
"O=pristine",
"R=agent said this",
"LI=19",
"LO=15"
]
);
assert!(hooks.drain_dispatch_faults().is_empty());
});
}
use crate::conversation_import::test_support::{
reader_double as conversation_reader_double, REPORT_SEP as CONVERSATION_SEP,
};
fn staged_conversation_hook(
component: Component,
conversation_read: bool,
) -> StagedHookArtifact {
StagedHookArtifact {
grant: HookCapabilityGrant {
conversation_read,
..HookCapabilityGrant::default()
},
..staged_double_named("reader", HookBinding::OnTaskEnd, component)
}
}
fn recorded_id(n: usize) -> String {
format!("msg_{n:032x}")
}
fn write_record(root: &Path, context_id: &str, count: usize) {
let dir = root.join(context_id);
std::fs::create_dir_all(&dir).unwrap();
let mut raw = String::new();
for n in 1..=count {
raw.push_str(
&serde_json::json!({
"role": "user",
"content": [{"type": "text", "text": format!("m{n}")}],
"id": recorded_id(n),
})
.to_string(),
);
raw.push('\n');
}
std::fs::write(dir.join("conversation.jsonl"), raw).unwrap();
}
fn read_pages(
root: Option<PathBuf>,
context_id: Option<&str>,
conversation_read: bool,
) -> Vec<String> {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged =
staged_conversation_hook(conversation_reader_double(&engine, 2), conversation_read);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = HookRuntime::new(
&engine,
session.path(),
accessible.path(),
vec![staged],
SessionContextData {
capsule_name: "test-capsule".to_string(),
capsule_version: "0.1.0".to_string(),
session_id: "sess-test".to_string(),
model: "test-model".to_string(),
capabilities: Vec::new(),
},
HookEnvVars::default(),
ExecutionLimits::default(),
None,
root,
)
.await
.expect("a conversation-importing hook instantiates like any other");
if let Some(context_id) = context_id {
hooks
.dispatch_task_start(
"tsk_1".to_string(),
context_id.to_string(),
"task_md".to_string(),
4,
0,
0,
0,
)
.await;
}
let reopen = hooks
.dispatch_task_end("tsk_1".to_string(), "ok".to_string())
.await
.expect("the reader double always returns reopen-task");
reopen.reason.split('|').map(str::to_string).collect()
})
}
#[test]
fn a_granted_hook_pages_a_prior_runs_record_newest_first() {
let home = TempDir::new().unwrap();
let root = home.path().join("conversations/test-capsule");
write_record(&root, "ctx_fixed", 5);
let pages = read_pages(Some(root), Some("ctx_fixed"), true);
assert_eq!(
pages[0],
format!(
"T=5{sep}{id5}=user{sep}{id4}=user{sep}N=mc_3",
sep = CONVERSATION_SEP,
id5 = recorded_id(5),
id4 = recorded_id(4),
),
"the newest two messages, each carrying the id its record line holds"
);
assert_eq!(
pages[1],
format!(
"T=5{sep}{id3}=user{sep}{id2}=user{sep}N=mc_1",
sep = CONVERSATION_SEP,
id3 = recorded_id(3),
id2 = recorded_id(2),
),
"the cursor continues where the first page stopped"
);
}
#[test]
fn the_final_page_reports_no_next_cursor() {
let home = TempDir::new().unwrap();
let root = home.path().join("conversations/test-capsule");
write_record(&root, "ctx_fixed", 2);
let pages = read_pages(Some(root), Some("ctx_fixed"), true);
assert!(pages[0].ends_with("N=-"), "page was: {}", pages[0]);
}
#[test]
fn an_ungranted_hook_reads_not_granted_rather_than_an_empty_page() {
let home = TempDir::new().unwrap();
let root = home.path().join("conversations/test-capsule");
write_record(&root, "ctx_fixed", 3);
let pages = read_pages(Some(root), Some("ctx_fixed"), false);
assert_eq!(pages, vec!["!not-granted", "!not-granted"]);
}
#[test]
fn a_session_with_no_record_reads_an_empty_page() {
assert_eq!(
read_pages(None, Some("ctx_fixed"), true),
vec![format!("T=0{CONVERSATION_SEP}N=-"); 2]
);
}
#[test]
fn a_malformed_record_line_is_skipped_and_the_read_still_succeeds() {
let home = TempDir::new().unwrap();
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let root = home.path().join("conversations/test-capsule");
write_record(&root, "ctx_fixed", 3);
let path = root.join("ctx_fixed/conversation.jsonl");
let mut raw = std::fs::read_to_string(&path).unwrap();
raw.push_str("{\"role\":\"user\"");
std::fs::write(&path, raw).unwrap();
let engine = hook_test_engine();
let staged = staged_conversation_hook(conversation_reader_double(&engine, 2), true);
let rt = tokio::runtime::Runtime::new().unwrap();
let reason = rt.block_on(async {
let mut hooks = HookRuntime::new(
&engine,
session.path(),
accessible.path(),
vec![staged],
SessionContextData {
capsule_name: "test-capsule".to_string(),
capsule_version: "0.1.0".to_string(),
session_id: "sess-test".to_string(),
model: "test-model".to_string(),
capabilities: Vec::new(),
},
HookEnvVars::default(),
ExecutionLimits::default(),
None,
Some(root),
)
.await
.unwrap();
hooks
.dispatch_task_start(
"tsk_1".to_string(),
"ctx_fixed".to_string(),
"task_md".to_string(),
4,
0,
0,
0,
)
.await;
hooks
.dispatch_task_end("tsk_1".to_string(), "ok".to_string())
.await
.expect("a malformed line never fails the read")
.reason
});
assert!(reason.starts_with("T=3"), "report was: {reason}");
let log = std::fs::read_to_string(session.path().join("logs/bootstrap.log")).unwrap();
assert!(
log.contains("line 4 is not a message"),
"the skipped line must be named: {log}"
);
}
use crate::network_policy::RequestTarget;
use murmur_artifact::{Capabilities, FilesystemCapabilities, NetworkCapabilities};
fn grant_of(network: Option<&str>, scope: Option<&str>) -> HookCapabilityGrant {
let caps = Capabilities {
peer_fetch: None,
network: network.map(|entry| NetworkCapabilities {
allow: vec![entry.to_string()],
unix_sockets: false,
}),
filesystem: scope.map(|scope| FilesystemCapabilities {
scope: Some(scope.to_string()),
workdir_exec: false,
read_only: Vec::new(),
}),
shell: None,
spawn: None,
env: None,
limits: None,
resources: None,
state: None,
task_io: None,
conversation: None,
plan: None,
containment: None,
};
HookCapabilityGrant::derive(Some(&caps), "test-capsule").expect("grant is valid")
}
fn hook_store_state(root: &Path, grant: &HookCapabilityGrant) -> HookStoreState {
HookStoreState {
limits: ExecutionLimits::default().limiter(),
table: ResourceTable::new(),
wasi: build_wasi_ctx(root, &HookEnvVars::default(), grant).expect("wasi ctx builds"),
http: WasiHttpCtx::new(),
http_hooks: NetworkPolicyHooks {
network_allow_rules: grant.network_allow_rules.clone(),
inference_gateway: None,
},
}
}
fn send_through_hook_store(state: &mut HookStoreState, uri: &str, use_tls: bool) -> bool {
use http_body_util::{BodyExt, Empty};
let body = Empty::<bytes::Bytes>::new()
.map_err(|err| match err {})
.boxed_unsync();
let request = hyper::Request::builder()
.uri(uri)
.body(body)
.expect("request builds");
let config = wasmtime_wasi_http::p2::types::OutgoingRequestConfig {
use_tls,
connect_timeout: Duration::from_millis(1),
first_byte_timeout: Duration::from_millis(1),
between_bytes_timeout: Duration::from_millis(1),
};
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async { state.http().hooks.send_request(request, config).is_ok() })
}
#[test]
fn ungranted_hook_store_denies_every_outbound_request() {
let root = TempDir::new().unwrap();
let grant = HookCapabilityGrant::default();
let mut state = hook_store_state(root.path(), &grant);
assert!(!send_through_hook_store(
&mut state,
"https://telemetry.example.com/ingest",
true
));
assert!(!send_through_hook_store(
&mut state,
"http://127.0.0.1:1/local",
false
));
}
#[test]
fn granted_hook_store_admits_only_the_declared_host() {
let root = TempDir::new().unwrap();
let grant = grant_of(Some("http://127.0.0.1:1"), None);
let mut state = hook_store_state(root.path(), &grant);
assert!(
send_through_hook_store(&mut state, "http://127.0.0.1:1/ingest", false),
"the granted host must pass the allow-list"
);
assert!(
!send_through_hook_store(&mut state, "http://127.0.0.1:2/ingest", false),
"a different port on the granted host must still be denied"
);
assert!(
!send_through_hook_store(&mut state, "https://evil.example.com/x", true),
"an undeclared host must be denied"
);
}
#[test]
fn ungranted_hook_gets_no_preopened_directory() {
let root = TempDir::new().unwrap();
let missing = root.path().join("does-not-exist");
build_wasi_ctx(
&missing,
&HookEnvVars::default(),
&HookCapabilityGrant::default(),
)
.expect("an ungranted hook preopens nothing, so a missing root is not an error");
assert!(
!missing.exists(),
"default-deny must not create the working directory either"
);
assert!(HookCapabilityGrant::default().state_dir.is_none());
}
#[test]
fn a_hook_granted_scope_and_state_preopens_both() {
let root = TempDir::new().unwrap();
let state = TempDir::new().unwrap();
std::fs::write(state.path().join("cursor.json"), b"{\"seen\":7}").unwrap();
let mut grant = grant_of(None, Some("hook-state"));
grant.state_store = Some("hook-store".to_string());
grant.state_dir = Some(state.path().to_path_buf());
build_wasi_ctx(root.path(), &HookEnvVars::default(), &grant)
.expect("both preopens are built");
assert!(root.path().join("hook-state").is_dir());
assert!(!state.path().starts_with(root.path()));
assert_eq!(
std::fs::read_to_string(state.path().join("cursor.json")).unwrap(),
"{\"seen\":7}"
);
}
#[test]
fn a_hook_granted_state_but_no_scope_preopens_only_the_store() {
let root = TempDir::new().unwrap();
let missing = root.path().join("does-not-exist");
let state = TempDir::new().unwrap();
let grant = HookCapabilityGrant {
state_store: Some("hook-store".to_string()),
state_dir: Some(state.path().to_path_buf()),
..HookCapabilityGrant::default()
};
build_wasi_ctx(&missing, &HookEnvVars::default(), &grant)
.expect("a state grant alone does not preopen the hook's working directory");
assert!(
!missing.exists(),
"a state grant must not create the working directory"
);
}
#[test]
fn a_missing_state_directory_fails_the_build() {
let root = TempDir::new().unwrap();
let absent = root.path().join("no-such-store");
let grant = HookCapabilityGrant {
state_store: Some("hook-store".to_string()),
state_dir: Some(absent.clone()),
..HookCapabilityGrant::default()
};
let err = match build_wasi_ctx(root.path(), &HookEnvVars::default(), &grant) {
Ok(_) => panic!("a missing store directory must fail the build"),
Err(err) => err,
};
assert!(
err.to_string().contains("no-such-store"),
"the error must name the path: {err}"
);
}
#[test]
fn granted_hook_preopens_only_the_scoped_subtree() {
let root = TempDir::new().unwrap();
std::fs::write(root.path().join("secret.txt"), b"capsule state").unwrap();
let sibling = root.path().join("other-artifact");
std::fs::create_dir_all(&sibling).unwrap();
let grant = grant_of(None, Some("hook-state"));
build_wasi_ctx(root.path(), &HookEnvVars::default(), &grant)
.expect("a granted scope is created and preopened");
let scoped = root.path().join("hook-state");
assert!(scoped.is_dir(), "the granted scope is created if missing");
std::fs::write(scoped.join("cursor.json"), b"{}").unwrap();
assert!(scoped.join("cursor.json").exists());
assert!(sibling.is_dir());
assert!(root.path().join("secret.txt").exists());
}
#[test]
fn granted_hook_scope_reuses_an_existing_directory() {
let root = TempDir::new().unwrap();
let scoped = root.path().join("hook-state");
std::fs::create_dir_all(&scoped).unwrap();
std::fs::write(scoped.join("cursor.json"), b"{\"seen\":7}").unwrap();
build_wasi_ctx(
root.path(),
&HookEnvVars::default(),
&grant_of(None, Some("hook-state")),
)
.unwrap();
assert_eq!(
std::fs::read_to_string(scoped.join("cursor.json")).unwrap(),
"{\"seen\":7}"
);
}
#[test]
fn unusable_scope_fails_instantiation() {
let root = TempDir::new().unwrap();
std::fs::write(root.path().join("hook-state"), b"not a directory").unwrap();
let err = match build_wasi_ctx(
root.path(),
&HookEnvVars::default(),
&grant_of(None, Some("hook-state")),
) {
Ok(_) => panic!("an uncreatable scope must be a hard error"),
Err(err) => err,
};
assert!(
err.to_string().contains("hook-state"),
"the error must name the scope: {err}"
);
}
#[test]
fn blocking_hook_instantiation_applies_the_grant() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let component = hook_double(&engine, &REQUIRED_HOOK_FNS);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_granted(
component,
grant_of(None, Some("hook-state")),
)],
)
.await
.expect("a granted blocking hook instantiates");
});
assert!(
accessible.path().join("hook-state").is_dir(),
"the blocking path preopens the scope under the project dir"
);
assert!(
!session.path().join("hook-state").exists(),
"the session dir is not the blocking hook's root and must stay untouched"
);
}
#[test]
fn ungranted_blocking_hook_instantiates_without_any_preopen() {
let session = TempDir::new().unwrap();
let engine = hook_test_engine();
let component = hook_double(&engine, &REQUIRED_HOOK_FNS);
let missing_project_dir = session.path().join("no-such-project");
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
new_with_hooks(
&engine,
session.path(),
&missing_project_dir,
vec![staged_double(component)],
)
.await
.expect("an ungranted blocking hook preopens nothing");
});
assert!(!missing_project_dir.exists());
}
#[test]
fn on_stage_instantiation_applies_the_grant() {
let workdir = TempDir::new().unwrap();
let engine = hook_test_engine();
let mut names: Vec<&str> = REQUIRED_HOOK_FNS.to_vec();
names.push("on-stage");
let staged = staged_double_granted(
hook_double(&engine, &names),
grant_of(None, Some("hook-state")),
);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let _ = call_stage_once(
&engine,
workdir.path(),
&staged,
&StageEvent {
shell_allow: Vec::new(),
},
&HookEnvVars::default(),
ExecutionLimits::default(),
)
.await;
});
assert!(
workdir.path().join("hook-state").is_dir(),
"the on-stage path preopens the scope under the session workdir"
);
}
#[test]
fn async_hook_instantiation_applies_the_grant() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_async_granted(
"async-hook",
HookBinding::All,
hook_double(&engine, &REQUIRED_HOOK_FNS),
grant_of(None, Some("hook-state")),
HookOverflowPolicy::Drop,
);
run_local(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("a granted async hook instantiates");
hooks.drain_async_hooks().await;
});
assert!(
accessible.path().join("hook-state").is_dir(),
"the async path preopens the scope like the other two"
);
assert!(
!session.path().join("hook-state").exists(),
"the session dir is not an async hook's root and must stay untouched"
);
}
#[test]
fn hook_self_declared_capabilities_do_not_grant_anything() {
let hook_own_manifest = r#"
name: greedy-hook
version: 1.0.0
binding: on-inference
capabilities:
network:
allow:
- https://evil.example.com
filesystem:
scope: .
"#;
let config = murmur_artifact::parse_hook_config_from_yaml(hook_own_manifest)
.expect("the hook's own manifest still parses for its behavioral contract");
assert_eq!(config.binding, HookBinding::OnInference);
let operator_manifest = murmur_artifact::RuntimeManifest::from_yaml_str(
r#"
name: cap
version: 0.0.1
artifacts:
- name: greedy-hook
version: 1.0.0
runtime: hook
"#,
)
.unwrap();
let grant = HookCapabilityGrant::derive(
operator_manifest.artifacts[0].capabilities.as_ref(),
"test-capsule",
)
.unwrap();
assert_eq!(
grant,
HookCapabilityGrant::default(),
"only the operator's entry may grant; the hook's own manifest is never read"
);
let root = TempDir::new().unwrap();
let mut state = hook_store_state(root.path(), &grant);
assert!(!send_through_hook_store(
&mut state,
"https://evil.example.com/exfil",
true
));
let target = RequestTarget {
scheme: "https".to_string(),
host: "evil.example.com".to_string(),
port: Some(443),
};
assert!(!grant
.network_allow_rules
.iter()
.any(|rule| rule.matches(&target)));
}
fn hook_tool_call_arm_double(engine: &wasmtime::Engine, arm_disc: i32) -> Component {
let stubs = REQUIRED_HOOK_FNS
.iter()
.filter(|n| **n != "on-tool-call")
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let iface = LIFECYCLE_IFACE;
let wat = format!(
r#"(component
(core module $m
(memory (export "memory") 1)
(func (export "realloc") (param i32 i32 i32 i32) (result i32) i32.const 512)
(func (export "ontool")
(param i32 i32 i32 i64 i32 i32 i32 i64 i64 i32 i32) (result i32)
(i32.store (i32.const 128) (i32.const 0))
(i32.store (i32.const 132) (i32.const {arm_disc}))
(i32.store (i32.const 136) (i32.const 0))
(i32.store (i32.const 140) (i32.const 0))
(i32.const 128))
(func (export "noop"))
)
(core instance $i (instantiate $m))
(alias core export $i "memory" (core memory $mem))
(alias core export $i "realloc" (core func $realloc))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $tool-outcome (record
(field "output-bytes" u64)
(field "duration-ms" u64)
(field "status" string)))
(type $tool-event (record
(field "turn" u32)
(field "tool-name" string)
(field "input-bytes" u64)
(field "input" string)
(field "outcome" (option $tool-outcome))))
(type $ft (func (param "event" $tool-event) (result (result $hook-output (error string)))))
(func $ot (type $ft)
(canon lift (core func $i "ontool") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "tool-outcome" (type $tool-outcome))
(export "tool-event" (type $tool-event))
(export "on-tool-call" (func $ot))
{stubs}
)
(export "{iface}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("tool-call component WAT parses");
Component::new(engine, &bytes).expect("tool-call component double compiles")
}
fn hook_task_end_arm_double(
engine: &wasmtime::Engine,
arm_disc: i32,
reason: &str,
) -> Component {
let stubs = REQUIRED_HOOK_FNS
.iter()
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let reason_len = reason.len();
let iface = LIFECYCLE_IFACE;
let wat = format!(
r#"(component
(core module $m
(memory (export "memory") 1)
(data (i32.const 300) "{reason}")
(func (export "realloc") (param i32 i32 i32 i32) (result i32) i32.const 512)
(func (export "ontaskend") (param i32 i32 i32 i32) (result i32)
(i32.store (i32.const 128) (i32.const 0))
(i32.store (i32.const 132) (i32.const {arm_disc}))
(i32.store (i32.const 136) (i32.const 300))
(i32.store (i32.const 140) (i32.const {reason_len}))
(i32.const 128))
(func (export "noop"))
)
(core instance $i (instantiate $m))
(alias core export $i "memory" (core memory $mem))
(alias core export $i "realloc" (core func $realloc))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $task-end-event (record
(field "task-id" string)
(field "exit-status" string)))
(type $ft (func (param "event" $task-end-event) (result (result $hook-output (error string)))))
(func $te (type $ft)
(canon lift (core func $i "ontaskend") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "task-end-event" (type $task-end-event))
(export "on-task-end" (func $te))
{stubs}
)
(export "{iface}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("task-end component WAT parses");
Component::new(engine, &bytes).expect("task-end component double compiles")
}
#[test]
fn on_task_end_reopen_task_is_surfaced_by_dispatch_task_end() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let component = hook_task_end_arm_double(&engine, 4, "tests still fail");
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"gatekeeper",
HookBinding::OnTaskEnd,
component,
)],
)
.await
.expect("task-end double instantiates");
let reopen = hooks
.dispatch_task_end("tsk_1".to_string(), "ok".to_string())
.await
.expect("on-task-end returned reopen-task, so a TaskReopen must surface");
assert_eq!(reopen.hook_name, "gatekeeper");
assert_eq!(reopen.reason, "tests still fail");
assert!(
hooks.drain_dispatch_faults().is_empty(),
"reopen-task is honored at on-task-end, not a fault"
);
});
assert_eq!(
hook_log_lines(session.path(), "gatekeeper"),
0,
"an honored reopen-task must not log a fault"
);
}
#[test]
fn on_task_end_none_surfaces_no_reopen() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let component = hook_task_end_arm_double(&engine, 0, "unused");
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"observer",
HookBinding::OnTaskEnd,
component,
)],
)
.await
.expect("task-end double instantiates");
let reopen = hooks
.dispatch_task_end("tsk_1".to_string(), "ok".to_string())
.await;
assert!(
reopen.is_none(),
"none from on-task-end must not request a reopen"
);
assert!(hooks.drain_dispatch_faults().is_empty());
});
}
#[test]
fn reopen_task_is_only_honored_at_on_task_end() {
for (event, arm) in HONORED_OUTPUT_ARM {
let honors_reopen = *arm == Some("reopen-task");
assert_eq!(
honors_reopen,
*event == "on-task-end",
"{event} honoring reopen-task must be exactly on-task-end"
);
}
}
fn hook_task_start_double(
engine: &wasmtime::Engine,
expect_window: u64,
arm_disc: i32,
) -> Component {
let stubs = REQUIRED_HOOK_FNS
.iter()
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let iface = LIFECYCLE_IFACE;
let wat = format!(
r#"(component
(core module $m
(memory (export "memory") 1)
(data (i32.const 300) "user")
(data (i32.const 320) "seed")
(func (export "realloc") (param i32 i32 i32 i32) (result i32) i32.const 512)
(func (export "ontaskstart")
(param $tid-ptr i32) (param $tid-len i32)
(param $cid-ptr i32) (param $cid-len i32)
(param $src-ptr i32) (param $src-len i32)
(param $input-bytes i64) (param $budget i64)
(param $window i64) (param $prior i64)
(result i32)
(if (i64.ne (local.get $window) (i64.const {expect_window})) (then unreachable))
(i32.store (i32.const 128) (i32.const 0))
(i32.store (i32.const 132) (i32.const {arm_disc}))
(i32.store (i32.const 136) (i32.const 160))
(i32.store (i32.const 140) (i32.const 1))
(i32.store (i32.const 160) (i32.const 300))
(i32.store (i32.const 164) (i32.const 4))
(i32.store (i32.const 168) (i32.const 320))
(i32.store (i32.const 172) (i32.const 4))
(i32.store (i32.const 176) (i32.const 0))
(i32.store (i32.const 188) (i32.const 0))
(i32.const 128))
(func (export "noop"))
)
(core instance $i (instantiate $m))
(alias core export $i "memory" (core memory $mem))
(alias core export $i "realloc" (core func $realloc))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $task-start-event (record
(field "task-id" string)
(field "context-id" string)
(field "source" string)
(field "input-bytes" u64)
(field "budget-tokens" u64)
(field "context-window" u64)
(field "prior-tokens" u64)))
(type $ft (func (param "event" $task-start-event) (result (result $hook-output (error string)))))
(func $ts (type $ft)
(canon lift (core func $i "ontaskstart") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "task-start-event" (type $task-start-event))
(export "on-task-start" (func $ts))
{stubs}
)
(export "{iface}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("task-start component WAT parses");
Component::new(engine, &bytes).expect("task-start component double compiles")
}
async fn dispatch_task_start_against(
session: &Path,
accessible: &Path,
engine: &wasmtime::Engine,
double: Component,
context_window: u64,
) -> Vec<DispatchFault> {
let mut hooks = new_with_hooks(
engine,
session,
accessible,
vec![staged_double_named(
"starter",
HookBinding::OnTaskStart,
double,
)],
)
.await
.expect("task-start double instantiates");
hooks
.emit(
session,
HookEvent::TaskStart {
task_id: "tsk_1".to_string(),
context_id: "ctx_1".to_string(),
source: "task_md".to_string(),
input_bytes: 12,
budget_tokens: 0,
context_window,
prior_tokens: 0,
},
)
.await;
hooks.drain_dispatch_faults()
}
async fn seed_from_task_start_against(
session: &Path,
accessible: &Path,
engine: &wasmtime::Engine,
double: Component,
context_window: u64,
) -> (Option<HookSeed>, Vec<DispatchFault>) {
let mut hooks = new_with_hooks(
engine,
session,
accessible,
vec![staged_double_named(
"starter",
HookBinding::OnTaskStart,
double,
)],
)
.await
.expect("task-start double instantiates");
let seed = hooks
.dispatch_task_start(
"tsk_1".to_string(),
"ctx_1".to_string(),
"task_md".to_string(),
12,
20_000,
context_window,
0,
)
.await;
(seed, hooks.drain_dispatch_faults())
}
#[test]
fn on_task_start_carries_the_context_window_to_the_guest() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let faults = rt.block_on(dispatch_task_start_against(
session.path(),
accessible.path(),
&engine,
hook_task_start_double(&engine, 200_000, 0),
200_000,
));
assert!(faults.is_empty(), "none is silent from every event");
assert_eq!(
hook_log_lines(session.path(), "starter"),
0,
"the guest must receive the exact context-window the host sent"
);
}
#[test]
fn on_task_start_carries_a_zero_context_window_when_the_manifest_declares_none() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let faults = rt.block_on(dispatch_task_start_against(
session.path(),
accessible.path(),
&engine,
hook_task_start_double(&engine, 0, 0),
0,
));
assert!(faults.is_empty());
assert_eq!(hook_log_lines(session.path(), "starter"), 0);
}
#[test]
fn seed_context_from_on_task_start_is_honored() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let (seed, faults) = rt.block_on(seed_from_task_start_against(
session.path(),
accessible.path(),
&engine,
hook_task_start_double(&engine, 0, 5),
0,
));
let seed = seed.expect("on-task-start must surface the hook's seed-context");
assert_eq!(seed.hook_name, "starter");
assert_eq!(seed.messages.len(), 1);
assert_eq!(seed.messages[0].role, "user");
assert_eq!(seed.messages[0].content, "seed");
assert!(
faults.is_empty(),
"an honored arm raises no fault: {faults:?}"
);
assert_eq!(hook_log_lines(session.path(), "starter"), 0);
}
#[test]
fn none_from_on_task_start_proposes_no_seed() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
let (seed, faults) = rt.block_on(seed_from_task_start_against(
session.path(),
accessible.path(),
&engine,
hook_task_start_double(&engine, 0, 0),
0,
));
assert!(seed.is_none());
assert!(faults.is_empty());
}
#[test]
fn only_on_task_start_honors_seed_context() {
for (event, arm) in HONORED_OUTPUT_ARM {
assert_eq!(
*arm == Some("seed-context"),
*event == "on-task-start",
"{event} honoring seed-context must be exactly on-task-start"
);
}
}
fn hook_stage_arm_double(engine: &wasmtime::Engine, arm_disc: i32) -> Component {
let stubs = REQUIRED_HOOK_FNS
.iter()
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let iface = LIFECYCLE_IFACE;
let wat = format!(
r#"(component
(core module $m
(memory (export "memory") 1)
(func (export "realloc") (param i32 i32 i32 i32) (result i32) i32.const 512)
(func (export "onstage") (param i32 i32) (result i32)
(i32.store (i32.const 128) (i32.const 0))
(i32.store (i32.const 132) (i32.const {arm_disc}))
(i32.store (i32.const 136) (i32.const 0))
(i32.store (i32.const 140) (i32.const 0))
(i32.const 128))
(func (export "noop"))
)
(core instance $i (instantiate $m))
(alias core export $i "memory" (core memory $mem))
(alias core export $i "realloc" (core func $realloc))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $stage-event (record (field "shell-allow" (list string))))
(type $ft (func (param "event" $stage-event) (result (result $hook-output (error string)))))
(func $os (type $ft)
(canon lift (core func $i "onstage") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "stage-event" (type $stage-event))
(export "on-stage" (func $os))
{stubs}
)
(export "{iface}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("stage component WAT parses");
Component::new(engine, &bytes).expect("stage component double compiles")
}
fn tool_call_event() -> HookEvent {
HookEvent::ToolCall {
turn: 0,
tool_name: "bash".to_string(),
input_bytes: 0,
input: String::new(),
output_bytes: 0,
duration_ms: 0,
status: "ok".to_string(),
}
}
#[test]
fn unsupported_arm_from_blocking_hook_is_logged_and_buffered() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let component = hook_tool_call_arm_double(&engine, 2);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"toolhook",
HookBinding::OnToolCall,
component,
)],
)
.await
.expect("double instantiates");
let artifacts = hooks.emit(session.path(), tool_call_event()).await;
assert!(
artifacts.is_empty(),
"an unsupported arm commits nothing, exactly as `none` would"
);
let faults = hooks.drain_dispatch_faults();
assert_eq!(faults.len(), 1, "exactly one buffered fault");
assert_eq!(faults[0].hook_name, "toolhook");
assert_eq!(faults[0].event, "on-tool-call");
assert_eq!(faults[0].arm, "write-manifests");
assert!(hooks.drain_dispatch_faults().is_empty());
});
assert_eq!(
hook_log_lines(session.path(), "toolhook"),
1,
"exactly one fault line is written to the per-hook log"
);
let log =
std::fs::read_to_string(session.path().join("logs").join("hook-toolhook.log")).unwrap();
assert!(log.contains("on-tool-call"), "log names the event: {log}");
assert!(log.contains("write-manifests"), "log names the arm: {log}");
assert!(log.contains("toolhook"), "log names the hook: {log}");
}
#[test]
fn none_from_blocking_hook_produces_no_fault() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let component = hook_tool_call_arm_double(&engine, 0);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"quiet",
HookBinding::OnToolCall,
component,
)],
)
.await
.expect("double instantiates");
hooks.emit(session.path(), tool_call_event()).await;
assert!(
hooks.drain_dispatch_faults().is_empty(),
"`none` must never buffer a fault"
);
});
assert_eq!(
hook_log_lines(session.path(), "quiet"),
0,
"`none` must never write a log line"
);
}
#[test]
fn honored_inference_artifact_produces_no_fault() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let component = hook_inference_caller_double(&engine);
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"inf",
HookBinding::OnInference,
component,
)],
)
.await
.expect("double instantiates");
let artifacts = hooks.emit(session.path(), inference_event()).await;
assert_eq!(artifacts.len(), 1, "the honored artifact arm is committed");
assert!(
hooks.drain_dispatch_faults().is_empty(),
"the honored arm must not be a fault"
);
});
assert_eq!(
hook_log_lines(session.path(), "inf"),
0,
"the honored arm must not write a fault line"
);
}
#[test]
fn honored_compaction_replace_context_produces_no_fault() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let mut hooks = new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged_double_named(
"compactor",
HookBinding::OnCompaction,
hook_compaction_echo_model_double(&engine),
)],
)
.await
.expect("double instantiates");
let result = hooks
.dispatch_compaction(
vec![Message {
role: "user".to_string(),
content: "hello".to_string(),
id: None,
source_id: None,
}],
1234,
0.98,
Some("claude-haiku-4-5".to_string()),
None,
)
.await;
let messages = result.expect("hook succeeded").expect("replace-context");
assert_eq!(messages[0].content, "claude-haiku-4-5");
assert!(
hooks.drain_dispatch_faults().is_empty(),
"the honored replace-context arm must not be a fault"
);
});
assert_eq!(
hook_log_lines(session.path(), "compactor"),
0,
"the honored arm must not write a fault line"
);
}
#[test]
fn on_stage_unsupported_arm_is_logged_not_traced() {
let workdir = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_double_named(
"stager",
HookBinding::OnStage,
hook_stage_arm_double(&engine, 1),
);
dispatch_stage(
&engine,
workdir.path(),
std::slice::from_ref(&staged),
Vec::new(),
&HookEnvVars::default(),
ExecutionLimits::default(),
)
.expect("dispatch_stage returns Ok even when a hook returns an unsupported arm");
assert_eq!(
hook_log_lines(workdir.path(), "stager"),
1,
"the on-stage fault is logged once"
);
let log =
std::fs::read_to_string(workdir.path().join("logs").join("hook-stager.log")).unwrap();
assert!(log.contains("on-stage"), "log names the event: {log}");
assert!(log.contains("replace-context"), "log names the arm: {log}");
assert!(
!workdir.path().join("trace.jsonl").exists(),
"on-stage runs before the trace writer exists, so no trace is attempted"
);
}
#[test]
fn on_stage_write_manifests_remains_honored() {
let workdir = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_double_named(
"stager",
HookBinding::OnStage,
hook_stage_arm_double(&engine, 2),
);
dispatch_stage(
&engine,
workdir.path(),
std::slice::from_ref(&staged),
Vec::new(),
&HookEnvVars::default(),
ExecutionLimits::default(),
)
.expect("dispatch_stage succeeds");
assert_eq!(
hook_log_lines(workdir.path(), "stager"),
0,
"the honored write-manifests arm must not log a fault"
);
}
#[test]
fn async_hook_unsupported_arm_is_logged_and_traced() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_async(
"async-tool",
HookBinding::OnToolCall,
hook_tool_call_arm_double(&engine, 2),
);
let faults = run_local(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("async double instantiates");
hooks.emit(session.path(), tool_call_event()).await;
hooks.drain_async_hooks().await;
hooks.drain_dispatch_faults()
});
assert_eq!(
hook_log_lines(session.path(), "async-tool"),
1,
"the async unsupported-arm fault is logged exactly once"
);
assert_eq!(faults.len(), 1, "and reaches the trace path exactly once");
assert_eq!(faults[0].hook_name, "async-tool");
assert_eq!(faults[0].event, "on-tool-call");
assert_eq!(faults[0].arm, "write-manifests");
}
#[test]
fn honored_output_arm_table_is_complete_and_consistent() {
let mut names: Vec<&str> = REQUIRED_HOOK_FNS.to_vec();
names.extend_from_slice(&OPTIONAL_HOOK_FNS);
names.push("on-stage");
assert_eq!(
HONORED_OUTPUT_ARM.len(),
names.len(),
"one honored-arm entry per lifecycle function"
);
for name in &names {
assert!(
HONORED_OUTPUT_ARM.iter().any(|(n, _)| n == name),
"{name} must have a honored-arm entry"
);
}
for (_, arm) in HONORED_OUTPUT_ARM {
if let Some(arm) = arm {
assert!(
matches!(
*arm,
"replace-context"
| "write-manifests"
| "artifact"
| "reopen-task"
| "seed-context"
),
"honored arm {arm} must be a real hook-output variant"
);
}
}
}
#[test]
fn honored_output_arm_agrees_with_manifest_commit_policy_for_binding() {
use murmur_artifact::HookCommitPolicy;
for binding in [
HookBinding::OnStage,
HookBinding::OnSessionStart,
HookBinding::OnTaskStart,
HookBinding::OnInference,
HookBinding::OnToolCall,
HookBinding::OnShell,
HookBinding::OnCompaction,
HookBinding::OnTaskEnd,
HookBinding::OnSessionEnd,
] {
let event = binding.as_str();
let arm = honored_arm(DispatchPhase::Observe, event);
let decision_arm = honored_arm(DispatchPhase::Decide, event);
let policy = murmur_artifact::commit_policy_for_binding(&binding);
if binding == HookBinding::OnInference {
assert_eq!(arm, Some("artifact"), "{event}");
assert_eq!(decision_arm, None, "{event}");
assert_eq!(policy, None, "{event}");
continue;
}
assert!(
arm.is_none() || decision_arm.is_none(),
"{event}: an event must honor an arm at one phase, not both"
);
let honored = arm.or(decision_arm);
assert_eq!(
honored,
policy.as_ref().map(HookCommitPolicy::as_str),
"{event}: honored arm and declarable commit_policy disagree"
);
}
assert_eq!(honored_arm(DispatchPhase::Decide, "on-shell"), Some("deny"));
assert_eq!(
honored_arm(DispatchPhase::Decide, "on-tool-call"),
Some("deny")
);
assert_eq!(honored_arm(DispatchPhase::Observe, "on-shell"), None);
assert_eq!(honored_arm(DispatchPhase::Observe, "on-tool-call"), None);
}
#[test]
fn deny_is_a_fault_at_every_observation_dispatch() {
for binding in [
HookBinding::OnStage,
HookBinding::OnSessionStart,
HookBinding::OnTaskStart,
HookBinding::OnInference,
HookBinding::OnToolCall,
HookBinding::OnShell,
HookBinding::OnCompaction,
HookBinding::OnTaskEnd,
HookBinding::OnSessionEnd,
] {
let event = binding.as_str();
let output = HookOutput::Deny("no".to_string());
assert!(
matches!(
classify_output(DispatchPhase::Observe, event, &output),
OutputDisposition::Fault("deny")
),
"{event}: deny must be a fault at the observation dispatch"
);
}
}
#[test]
fn only_deny_is_honored_at_a_decision_point() {
for event in ["on-shell", "on-tool-call"] {
assert!(matches!(
classify_output(
DispatchPhase::Decide,
event,
&HookOutput::Deny("no".to_string())
),
OutputDisposition::Honored
));
assert!(matches!(
classify_output(DispatchPhase::Decide, event, &HookOutput::None),
OutputDisposition::Ignore
));
assert!(matches!(
classify_output(
DispatchPhase::Decide,
event,
&HookOutput::Artifact("{}".to_string())
),
OutputDisposition::Fault("artifact")
));
}
}
#[test]
fn a_call_decision_has_two_variants_and_carries_no_grant() {
for decision in [
CallDecision::Proceed,
CallDecision::Denied {
hook_name: "policy".to_string(),
reason: "no".to_string(),
},
] {
match decision {
CallDecision::Proceed => {}
CallDecision::Denied { hook_name, reason } => {
assert_eq!(hook_name, "policy");
assert_eq!(reason, "no");
}
}
}
}
#[test]
fn gates_calls_is_true_only_for_a_bound_policy_hook() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let rt = tokio::runtime::Runtime::new().unwrap();
for (binding, commit_policy, expected) in [
(HookBinding::OnShell, HookCommitPolicy::Deny, true),
(HookBinding::OnToolCall, HookCommitPolicy::Deny, true),
(HookBinding::OnShell, HookCommitPolicy::None, false),
(HookBinding::OnToolCall, HookCommitPolicy::None, false),
(HookBinding::OnCompaction, HookCommitPolicy::Deny, false),
(HookBinding::All, HookCommitPolicy::None, false),
] {
let label = format!("{binding:?} with {commit_policy:?}");
let mut staged =
staged_double_named("probe", binding, hook_double(&engine, &REQUIRED_HOOK_FNS));
staged.config.commit_policy = commit_policy;
let hooks = rt
.block_on(new_with_hooks(
&engine,
session.path(),
accessible.path(),
vec![staged],
))
.expect("the probe double instantiates");
assert_eq!(hooks.gates_calls(), expected, "{label}");
}
}
fn run_local<F: std::future::Future>(body: F) -> F::Output {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("current-thread runtime builds");
let local = tokio::task::LocalSet::new();
local.block_on(&rt, body)
}
fn staged_async(name: &str, binding: HookBinding, component: Component) -> StagedHookArtifact {
staged_async_granted(
name,
binding,
component,
HookCapabilityGrant::default(),
HookOverflowPolicy::Drop,
)
}
fn staged_async_granted(
name: &str,
binding: HookBinding,
component: Component,
grant: HookCapabilityGrant,
on_overflow: HookOverflowPolicy,
) -> StagedHookArtifact {
StagedHookArtifact {
name: name.to_string(),
version: "0.0.1".to_string(),
component,
config: HookConfig {
binding,
execution_mode: HookExecutionMode::Async,
..HookConfig::default()
},
grant,
on_overflow,
}
}
const MARKER_BUFFER: i32 = 4096;
fn hook_marker_double(engine: &wasmtime::Engine) -> Component {
let stubs = REQUIRED_HOOK_FNS
.iter()
.filter(|n| !matches!(**n, "on-session-start" | "on-tool-call" | "on-session-end"))
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let iface = LIFECYCLE_IFACE;
let wat = format!(
r#"(component
(core module $m
(memory (export "memory") 1)
(global $n (mut i32) (i32.const 0))
(func (export "realloc") (param i32 i32 i32 i32) (result i32) i32.const 1024)
;; Append one marker byte and return ok(none): discriminant 0 (ok) at 128,
;; hook-output discriminant 0 (none) at 132.
(func $mark (param $ch i32)
(i32.store8 (i32.add (i32.const {MARKER_BUFFER}) (global.get $n)) (local.get $ch))
(global.set $n (i32.add (global.get $n) (i32.const 1))))
(func $ok (result i32)
(i32.store (i32.const 128) (i32.const 0))
(i32.store (i32.const 132) (i32.const 0))
(i32.const 128))
(func (export "onstart") (param i32 i32 i32 i32 i32 i32 i32 i32 i32 i32) (result i32)
(call $mark (i32.const 83))
(call $ok))
(func (export "ontool") (param i32 i32 i32 i64 i32 i32 i32 i64 i64 i32 i32) (result i32)
(call $mark (i32.const 84))
(call $ok))
;; err(<history>): discriminant 1 at 128, string (ptr,len) at 132/136.
(func (export "onend") (param i32 i64 i64 i32 i32 i64 i32 i32) (result i32)
(call $mark (i32.const 69))
(i32.store (i32.const 128) (i32.const 1))
(i32.store (i32.const 132) (i32.const {MARKER_BUFFER}))
(i32.store (i32.const 136) (global.get $n))
(i32.const 128))
(func (export "noop"))
)
(core instance $i (instantiate $m))
(alias core export $i "memory" (core memory $mem))
(alias core export $i "realloc" (core func $realloc))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $session-context (record
(field "capsule-name" string)
(field "capsule-version" string)
(field "session-id" string)
(field "model" string)
(field "capabilities" (list string))))
(type $tool-outcome (record
(field "output-bytes" u64)
(field "duration-ms" u64)
(field "status" string)))
(type $tool-event (record
(field "turn" u32)
(field "tool-name" string)
(field "input-bytes" u64)
(field "input" string)
(field "outcome" (option $tool-outcome))))
(type $session-end-event (record
(field "total-turns" u32)
(field "total-input-tokens" u64)
(field "total-output-tokens" u64)
(field "total-tool-calls" u32)
(field "total-shell-calls" u32)
(field "duration-ms" u64)
(field "exit-status" string)))
(type $fs (func (param "ctx" $session-context) (result (result $hook-output (error string)))))
(type $ft (func (param "event" $tool-event) (result (result $hook-output (error string)))))
(type $fe (func (param "event" $session-end-event) (result (result $hook-output (error string)))))
(func $os (type $fs)
(canon lift (core func $i "onstart") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $ot (type $ft)
(canon lift (core func $i "ontool") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $oe (type $fe)
(canon lift (core func $i "onend") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "session-context" (type $session-context))
(export "tool-outcome" (type $tool-outcome))
(export "tool-event" (type $tool-event))
(export "session-end-event" (type $session-end-event))
(export "on-session-start" (func $os))
(export "on-tool-call" (func $ot))
(export "on-session-end" (func $oe))
{stubs}
)
(export "{iface}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("marker component WAT parses");
Component::new(engine, &bytes).expect("marker component double compiles")
}
fn marker_history(session: &Path, hook_name: &str) -> String {
let path = session.join("logs").join(format!("hook-{hook_name}.log"));
let contents = std::fs::read_to_string(&path)
.unwrap_or_else(|e| panic!("{} must exist: {e}", path.display()));
contents.trim_end().to_string()
}
#[test]
fn async_hook_reuses_one_instance_across_events() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_async("stateful", HookBinding::All, hook_marker_double(&engine));
run_local(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("async double instantiates once, at new");
hooks.emit(session.path(), HookEvent::SessionStart).await;
hooks.emit(session.path(), tool_call_event()).await;
hooks
.emit(
session.path(),
HookEvent::SessionEnd {
total_turns: 1,
exit_status: "ok".to_string(),
},
)
.await;
hooks.drain_async_hooks().await;
});
assert_eq!(
marker_history(session.path(), "stateful"),
"STE",
"the second and third calls must see the state the first left behind"
);
}
#[test]
fn async_hook_calls_are_serialized_in_dispatch_order() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_async("ordered", HookBinding::All, hook_marker_double(&engine));
run_local(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("async double instantiates");
hooks.emit(session.path(), tool_call_event()).await;
hooks.emit(session.path(), HookEvent::SessionStart).await;
hooks.emit(session.path(), tool_call_event()).await;
hooks
.emit(
session.path(),
HookEvent::SessionEnd {
total_turns: 2,
exit_status: "ok".to_string(),
},
)
.await;
hooks.drain_async_hooks().await;
});
assert_eq!(
marker_history(session.path(), "ordered"),
"TSTE",
"the hook must observe exactly the dispatch order, with no interleaving"
);
}
fn hook_env_probe_double(engine: &wasmtime::Engine) -> Component {
let stubs = REQUIRED_HOOK_FNS
.iter()
.filter(|n| **n != "on-session-start")
.map(|n| format!(" (export \"{n}\" (func $noop))"))
.collect::<Vec<_>>()
.join("\n");
let iface = LIFECYCLE_IFACE;
let wat = format!(
r#"(component
(import "wasi:cli/environment@0.2.12" (instance $envi
(export "get-environment" (func (result (list (tuple string string)))))))
;; The memory must exist before `get-environment` can be lowered against it, and the
;; lowered function must exist before the module using it can be instantiated — hence
;; the split into an allocator module and a body module.
(core module $alloc
(memory (export "memory") 1)
(global $bump (mut i32) (i32.const 1024))
(data (i32.const 700) "no-env")
(func (export "realloc") (param i32 i32 i32 i32) (result i32)
(local $r i32)
(global.set $bump
(i32.and (i32.add (global.get $bump) (i32.const 7)) (i32.const -8)))
(local.set $r (global.get $bump))
(global.set $bump (i32.add (global.get $bump) (local.get 3)))
(local.get $r))
)
(core instance $a (instantiate $alloc))
(alias core export $a "memory" (core memory $mem))
(alias core export $a "realloc" (core func $realloc))
(core func $get-env
(canon lower (func $envi "get-environment")
(memory $mem) (realloc $realloc) string-encoding=utf8))
(core module $m
(import "host" "memory" (memory 1))
(import "host" "get-environment" (func $get_env (param i32)))
;; err(<first env value>): result discriminant 1 at 128, string (ptr,len) at 132/136.
(func (export "onstart") (param i32 i32 i32 i32 i32 i32 i32 i32 i32 i32) (result i32)
(local $base i32)
(call $get_env (i32.const 512))
(local.set $base (i32.load (i32.const 512)))
(i32.store (i32.const 128) (i32.const 1))
(if (i32.eqz (i32.load (i32.const 516)))
(then
(i32.store (i32.const 132) (i32.const 700))
(i32.store (i32.const 136) (i32.const 6)))
(else
(i32.store (i32.const 132)
(i32.load (i32.add (local.get $base) (i32.const 8))))
(i32.store (i32.const 136)
(i32.load (i32.add (local.get $base) (i32.const 12))))))
(i32.const 128))
(func (export "noop"))
)
(core instance $imports
(export "memory" (memory $mem))
(export "get-environment" (func $get-env))
)
(core instance $i (instantiate $m (with "host" (instance $imports))))
(type $message (record
(field "role" string)
(field "content" string)
(field "id" (option string))
(field "source-id" (option string))))
(type $tool-manifest (record (field "binary-name" string) (field "content" string)))
(type $hook-output (variant
(case "none")
(case "replace-context" (list $message))
(case "write-manifests" (list $tool-manifest))
(case "artifact" string)
(case "reopen-task" string)
(case "seed-context" (list $message))
(case "deny" string)))
(type $session-context (record
(field "capsule-name" string)
(field "capsule-version" string)
(field "session-id" string)
(field "model" string)
(field "capabilities" (list string))))
(type $fs (func (param "ctx" $session-context) (result (result $hook-output (error string)))))
(func $os (type $fs)
(canon lift (core func $i "onstart") (memory $mem) (realloc $realloc) string-encoding=utf8))
(func $noop (canon lift (core func $i "noop")))
(instance $lc
(export "message" (type $message))
(export "tool-manifest" (type $tool-manifest))
(export "hook-output" (type $hook-output))
(export "session-context" (type $session-context))
(export "on-session-start" (func $os))
{stubs}
)
(export "{iface}" (instance $lc))
)"#
);
let bytes = wat::parse_str(&wat).expect("env-probe component WAT parses");
Component::new(engine, &bytes).expect("env-probe component double compiles")
}
#[test]
fn async_hook_receives_the_session_environment() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_async(
"exporter",
HookBinding::OnSessionStart,
hook_env_probe_double(&engine),
);
run_local(async {
let mut hooks = new_with_hooks_full(
&engine,
session.path(),
accessible.path(),
vec![staged],
ExecutionLimits::default(),
HookEnvVars {
otel_endpoint: Some("http://example.test:4318"),
..HookEnvVars::default()
},
)
.await
.expect("env-probe double instantiates");
hooks.emit(session.path(), HookEvent::SessionStart).await;
hooks.drain_async_hooks().await;
});
assert_eq!(
marker_history(session.path(), "exporter"),
"http://example.test:4318",
"the async instance must carry the session's real MURMUR_OTEL_ENDPOINT"
);
}
fn first_hook_env_entry(grant: HookCapabilityGrant, env_vars: HookEnvVars<'_>) -> String {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_async_granted(
"probe",
HookBinding::OnSessionStart,
hook_env_probe_double(&engine),
grant,
HookOverflowPolicy::Drop,
);
run_local(async {
let mut hooks = new_with_hooks_full(
&engine,
session.path(),
accessible.path(),
vec![staged],
ExecutionLimits::default(),
env_vars,
)
.await
.expect("env-probe double instantiates");
hooks.emit(session.path(), HookEvent::SessionStart).await;
hooks.drain_async_hooks().await;
});
marker_history(session.path(), "probe")
}
#[test]
fn a_hook_granted_config_sees_it_in_its_environment() {
let grant = HookCapabilityGrant {
config_json: Some(r#"{"who":"hook"}"#.to_string()),
..HookCapabilityGrant::default()
};
assert_eq!(
first_hook_env_entry(grant, HookEnvVars::default()),
r#"{"who":"hook"}"#
);
}
#[test]
fn an_unconfigured_hook_sees_no_environment_at_all() {
assert_eq!(
first_hook_env_entry(HookCapabilityGrant::default(), HookEnvVars::default()),
"no-env"
);
}
#[test]
fn config_is_injected_before_the_runtime_owned_variables() {
let grant = HookCapabilityGrant {
config_json: Some(r#"{"who":"hook"}"#.to_string()),
..HookCapabilityGrant::default()
};
assert_eq!(
first_hook_env_entry(
grant,
HookEnvVars {
otel_endpoint: Some("http://example.test:4318"),
..HookEnvVars::default()
},
),
r#"{"who":"hook"}"#,
"the config must precede MURMUR_OTEL_ENDPOINT in the guest environment"
);
}
#[test]
fn async_hook_session_end_work_survives_local_set_teardown() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_async("exporter", HookBinding::All, hook_marker_double(&engine));
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
rt.block_on(async {
let local = tokio::task::LocalSet::new();
local
.run_until(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("async double instantiates");
hooks.emit(session.path(), HookEvent::SessionStart).await;
hooks.emit(session.path(), tool_call_event()).await;
hooks
.emit(
session.path(),
HookEvent::SessionEnd {
total_turns: 1,
exit_status: "ok".to_string(),
},
)
.await;
hooks.drain_async_hooks().await;
})
.await;
});
assert_eq!(
marker_history(session.path(), "exporter"),
"STE",
"the session-end call must have completed, not been dropped with the LocalSet"
);
}
#[test]
fn async_hook_that_overruns_the_drain_budget_is_reported() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let _ticker = crate::limits::EpochTicker::spawn(&engine);
let limits = ExecutionLimits {
deadline_seconds: 1,
..ExecutionLimits::default()
};
let staged = staged_async(
"wedged",
HookBinding::OnSessionStart,
hook_spin_double(&engine),
);
let faults = run_local(async {
let mut hooks = new_with_hooks_limited(
&engine,
session.path(),
accessible.path(),
vec![staged],
limits,
)
.await
.expect("the spin double instantiates");
hooks.emit(session.path(), HookEvent::SessionStart).await;
hooks.emit(session.path(), HookEvent::SessionStart).await;
hooks
.drain_async_hooks_within(Duration::from_millis(50))
.await;
hooks.drain_dispatch_faults()
});
assert!(
faults.iter().any(|f| f.hook_name == "wedged"
&& f.event == FAULT_EVENT_DRAIN
&& f.arm == FAULT_ARM_TIMEOUT),
"the abandoned hook must reach the trace path: {faults:?}"
);
let log = std::fs::read_to_string(session.path().join("logs").join("hook-wedged.log"))
.expect("the drain must log what it abandoned");
assert!(
log.contains("drain budget"),
"the log must name the budget that ran out: {log}"
);
}
#[test]
fn async_hook_overflow_drop_discards_and_counts() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let surplus = 5;
let staged = staged_async(
"lossy",
HookBinding::OnToolCall,
hook_marker_double(&engine),
);
let (dropped, faults) = run_local(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("async double instantiates");
for _ in 0..(ASYNC_HOOK_QUEUE_DEPTH + surplus) {
hooks.emit(session.path(), tool_call_event()).await;
}
let dropped = hooks.async_hooks[0].dropped;
(dropped, hooks.drain_dispatch_faults())
});
assert_eq!(
dropped, surplus as u64,
"exactly the events past the queue's depth are discarded"
);
let overflow: Vec<_> = faults
.iter()
.filter(|f| f.arm == FAULT_ARM_QUEUE_OVERFLOW)
.collect();
assert_eq!(
overflow.len(),
surplus,
"every discarded event is reported, not just counted: {faults:?}"
);
assert_eq!(overflow[0].hook_name, "lossy");
assert_eq!(overflow[0].event, "on-tool-call");
}
#[test]
fn async_hook_overflow_block_applies_backpressure() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let surplus = 3;
let staged = staged_async_granted(
"lossless",
HookBinding::All,
hook_marker_double(&engine),
HookCapabilityGrant::default(),
HookOverflowPolicy::Block,
);
let dropped = run_local(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("async double instantiates");
for _ in 0..(ASYNC_HOOK_QUEUE_DEPTH + surplus) {
hooks.emit(session.path(), tool_call_event()).await;
}
hooks
.emit(
session.path(),
HookEvent::SessionEnd {
total_turns: 0,
exit_status: "ok".to_string(),
},
)
.await;
let dropped = hooks.async_hooks[0].dropped;
hooks.drain_async_hooks().await;
dropped
});
assert_eq!(dropped, 0, "block must never discard an event");
let history = marker_history(session.path(), "lossless");
assert_eq!(
history.len(),
ASYNC_HOOK_QUEUE_DEPTH + surplus + 1,
"every dispatched event must reach the hook"
);
assert!(history.ends_with('E'), "history: {}", &history[..8]);
}
#[test]
fn async_hook_error_reaches_the_trace_fault_path() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_async(
"failing",
HookBinding::OnSessionEnd,
hook_marker_double(&engine),
);
let faults = run_local(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("async double instantiates");
hooks
.emit(
session.path(),
HookEvent::SessionEnd {
total_turns: 0,
exit_status: "ok".to_string(),
},
)
.await;
hooks.drain_async_hooks().await;
hooks.drain_dispatch_faults()
});
assert_eq!(faults.len(), 1, "the error must be traced once: {faults:?}");
assert_eq!(faults[0].hook_name, "failing");
assert_eq!(faults[0].event, "on-session-end");
assert_eq!(faults[0].arm, FAULT_ARM_ERROR);
assert_eq!(
hook_log_lines(session.path(), "failing"),
1,
"and still be logged exactly once"
);
}
#[test]
fn async_hook_replace_context_is_never_applied() {
let session = TempDir::new().unwrap();
let accessible = TempDir::new().unwrap();
let engine = hook_test_engine();
let staged = staged_async(
"compactor",
HookBinding::OnCompaction,
hook_compaction_echo_model_double(&engine),
);
let (result, faults) = run_local(async {
let mut hooks =
new_with_hooks(&engine, session.path(), accessible.path(), vec![staged])
.await
.expect("async double instantiates");
let result = hooks
.dispatch_compaction(
vec![Message {
role: "user".to_string(),
content: "hello".to_string(),
id: None,
source_id: None,
}],
1234,
0.98,
Some("claude-haiku-4-5".to_string()),
None,
)
.await;
hooks.drain_async_hooks().await;
(result, hooks.drain_dispatch_faults())
});
assert!(
result
.expect("an async hook cannot fail the compaction")
.is_none(),
"an async hook's replace-context must never reach the conversation"
);
assert!(faults.is_empty(), "and is not a fault either: {faults:?}");
}
}