capsule-runtime 0.7.0

use std::{
    collections::HashMap,
    path::{Path, PathBuf},
    sync::Arc,
    time::{Duration, Instant},
};

use wasmtime::{component::Linker, Store};
use wasmtime_wasi::{FsPerms, ResourceTable, WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView};
use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpCtxView, WasiHttpView};

use murmur_artifact::{
    HookBinding, HookCommitPolicy, HookConfig, HookExecutionMode, HookOverflowPolicy,
    PACKED_MANIFEST_ENTRY,
};
use tokio::sync::mpsc;

use crate::workdir_fs::{WorkdirRoot, DEFAULT_FILE_MODE};
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, InferenceUnavailable,
    },
    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,
};

/// The one lifecycle instance export name the host accepts, matching the
/// `murmur:hook` version declared in `wit/`. The host keeps no compatibility
/// fallback: a hook compiled against any other version does not resolve, so a
/// WIT bump requires every hook artifact to be rebuilt (see `wit/VERSIONING.md`).
pub(crate) const LIFECYCLE_IFACE: &str = "murmur:hook/lifecycle@0.9.0";

/// Resolve the lifecycle instance export. `None` means the component does not
/// export [`LIFECYCLE_IFACE`], which surfaces as a missing-export error at the
/// call site.
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)
}

/// Diagnostic naming the accepted lifecycle export name, used wherever
/// resolution fails.
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)"
    )
}

/// Depth of every async hook's job queue.
///
/// Fixed in code rather than exposed as a manifest knob: the depth only decides how long a
/// burst may outrun a hook before `on_overflow` starts applying, and that policy — not the
/// buffer size — is the operator-facing choice. Generous enough that no realistic burst of
/// lifecycle events reaches it while a healthy hook is keeping up; a queue that fills has a
/// hook that is genuinely too slow, which is a fact to report, not to buffer harder.
const ASYNC_HOOK_QUEUE_DEPTH: usize = 1024;

/// Total budget for [`HookRuntime::drain_async_hooks`], shared across every async hook.
///
/// Bounds how long session teardown may wait for hooks that are still working — long enough
/// for a queued session-end export to finish an HTTP round trip, short enough that a wedged
/// hook cannot hold the process open. Whatever a hook has not finished by then is abandoned
/// and reported (see [`FAULT_ARM_TIMEOUT`]).
const ASYNC_HOOK_DRAIN_TIMEOUT: Duration = Duration::from_secs(15);

/// [`DispatchFault::event`] for a fault raised by the session-end drain rather than by one
/// lifecycle call — there is no single event to name when a whole queue ran out of time.
const FAULT_EVENT_DRAIN: &str = "drain";

/// [`DispatchFault::arm`] for a hook call that returned a genuine `Err`, which has no
/// `hook-output` arm to name.
const FAULT_ARM_ERROR: &str = "error";

/// [`DispatchFault::arm`] for a lifecycle event discarded because an async hook's queue was
/// full and its entry declares `on_overflow: drop`.
const FAULT_ARM_QUEUE_OVERFLOW: &str = "queue-overflow";

/// [`DispatchFault::arm`] for an async hook that was still working when the drain budget ran
/// out.
const FAULT_ARM_TIMEOUT: &str = "timeout";

/// [`DispatchFault::arm`] for a `seed-context` the runtime honored the arm of but refused to
/// commit — no budget to enforce it against, a single message wider than the whole budget, an
/// overflow several multiples past it, or a transport with no message list to seed. The arm
/// was honored, so this is not an unsupported-arm fault; it is recorded beside the
/// `context_seed` event so a rejection is visible in `trace.jsonl` without reading
/// `logs/bootstrap.log`.
pub(crate) const FAULT_ARM_SEED_REJECTED: &str = "seed-rejected";

pub(crate) struct HookRuntime {
    /// The session's records directory, which every hook fault and `logs/hook-<name>.log` line
    /// is written through.
    records_root: WorkdirRoot,
    blocking_hooks: Vec<HookInstance>,
    /// One entry per `execution_mode: async` hook: its queue and the worker draining it. The
    /// instance itself lives inside that worker, instantiated once at [`Self::new`] exactly
    /// like a blocking hook's, so hook-side state survives across events.
    async_hooks: Vec<AsyncHookQueue>,
    /// Faults raised off the `&mut self` dispatch path — by an async hook's worker task, or by
    /// dispatch itself when a queue overflows. A channel rather than a direct push because a
    /// worker runs concurrently with (and outlives the borrow of) the dispatch loop; folded
    /// into [`Self::drain_dispatch_faults`] alongside [`Self::dispatch_faults`].
    fault_tx: mpsc::UnboundedSender<DispatchFault>,
    fault_rx: mpsc::UnboundedReceiver<DispatchFault>,
    context: SessionContextData,
    /// Backing for the `murmur:runtime/inference` host import. `Err` when the
    /// capsule has no inference driver a hook can call — `run-inference` then
    /// returns the reason's `err` instead of the import failing to link.
    inference: Result<Arc<HookInferenceCtx>, InferenceUnavailable>,
    /// Backing for the `murmur:task-io/read` host import: the runtime's single copy of the
    /// in-scope task's input and result text. Held for the whole launch and shared by `Arc`
    /// with each *granted* hook's linker, so what [`Self::begin_task_attempt`] and
    /// [`Self::record_task_output`] put here is what a hook call reads. A hook without
    /// `capabilities.task_io.read` never receives the `Arc`.
    task_io: Arc<TaskIoState>,
    /// The conversation record in scope, shared with every hook granted
    /// `capabilities.conversation.read` on the same terms as [`Self::task_io`].
    conversation: Arc<ConversationState>,
    /// Unsupported-arm faults produced by blocking hooks since the last drain, in
    /// dispatch order, plus drain-timeout faults. Drained by the agent loop via
    /// [`Self::drain_dispatch_faults`] and written to `trace.jsonl` as `hook_dispatch_error`
    /// events before each `session_end` write. Mirrors the drain idiom used for
    /// `run-inference` records (see [`Self::drain_inference_records`]). Faults raised off
    /// this path arrive through [`Self::fault_tx`] instead.
    dispatch_faults: Vec<DispatchFault>,
    /// Whether any staged blocking hook is a *policy* hook — `commit_policy: deny` bound to
    /// `on-shell` or `on-tool-call`. Computed once at [`HookRuntime::new`] and read as a single
    /// boolean by the agent loop before every tool call, so a capsule with no policy hook does
    /// no call resolution and no extra dispatch at all.
    gates_calls: bool,
    started: Instant,
    total_input_tokens: u64,
    total_output_tokens: u64,
    total_tool_calls: u32,
    total_shell_calls: u32,
    total_turns: u32,
}

/// The six `murmur:hook/lifecycle` functions every compiled hook component must export;
/// a missing one is a hard, session-fatal error at instantiation time.
const REQUIRED_HOOK_FNS: [&str; 6] = [
    "on-session-start",
    "on-inference",
    "on-tool-call",
    "on-shell",
    "on-compaction",
    "on-session-end",
];

/// Lifecycle functions introduced after the original six. A hook component compiled
/// before these existed may omit them; it is simply not dispatched for those events.
const OPTIONAL_HOOK_FNS: [&str; 2] = ["on-task-start", "on-task-end"];

/// Single source of truth for which `hook-output` arm each of the nine lifecycle
/// events honors at its [`DispatchPhase::Observe`] dispatch. Keyed by the WIT function
/// name; the value is the one non-`none` arm the runtime commits for that event, or
/// `None` when the event honors nothing beyond the always-silent `none`.
///
/// `deny` appears in no row here, and that is what makes it a fault from every observation
/// dispatch — including the observation dispatch of the two events that *do* honor it before
/// the call. See [`HONORED_DECISION_ARM`] for that phase.
///
/// Both dispatch paths consult this table — the shared [`call_hook`] path (for the
/// eight events routed through [`HookRuntime::dispatch`]) and the separate
/// [`dispatch_stage`]/[`call_stage_once`] path (for `on-stage`) — so the
/// honored-arm decision lives in exactly one place. Any non-`none` arm a hook
/// returns that is *not* this event's honored arm is a loud, non-fatal
/// unsupported-arm fault (logged, and for the shared path also traced); see
/// [`classify_output`].
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),
];

/// The honored-arm table for the *decision-point* dispatch, the sibling of
/// [`HONORED_OUTPUT_ARM`] and deliberately separate from it.
///
/// Only the two gated events appear, and only with `deny`. The seven other events have no
/// decision-point dispatch at all, so "a hook returning `deny` from anything else is a fault"
/// is a consequence of the tables rather than a special case written anywhere: `deny` is
/// absent from every row of [`HONORED_OUTPUT_ARM`], and every event but these two is absent
/// from this table.
const HONORED_DECISION_ARM: &[(&str, Option<&str>)] =
    &[("on-tool-call", Some("deny")), ("on-shell", Some("deny"))];

/// Characters of `shell-event.command` a hook is shown. Both dispatches of `on-shell` clip to
/// the same width, so a hook cannot tell the decision point from the observation by the length
/// of a display string. `argv` and `script` carry the untruncated call, and are what a policy
/// must decide on — the WIT doc comment on `command` says so.
const SHELL_COMMAND_DISPLAY_CHARS: usize = 200;

/// Which of an event's two dispatches this is.
///
/// The two gated events are dispatched twice over their lifetime — once to decide, once to
/// observe — and the two phases honor different arms, so the honored-arm decision is keyed on
/// the phase as well as the event name. Every other event has an [`Observe`](Self::Observe)
/// dispatch only.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum DispatchPhase {
    /// After the call, reporting what happened. The event carries `outcome: some(...)`, and
    /// nothing a hook returns can change what already ran.
    Observe,
    /// Before the call, deciding whether it happens. The event carries `outcome: none`.
    Decide,
}

/// The `hook-output` arm `event` (a WIT lifecycle function name) honors at `phase` beyond the
/// always-silent `none`, or `None` if it honors nothing else. Looks the event up in
/// [`HONORED_OUTPUT_ARM`] or [`HONORED_DECISION_ARM`] as the phase selects; an event absent
/// from the phase's table also yields `None`.
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)
}

/// The `hook-output` variant name a hook returned, or `None` for the `none` arm.
/// The returned `&'static str` values are the same arm spellings used in
/// [`HONORED_OUTPUT_ARM`] and in the WIT `variant hook-output`.
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"),
    }
}

/// What [`call_hook`] / [`dispatch_stage`] should do with one hook's returned
/// `hook-output` for a given event, decided solely against [`HONORED_OUTPUT_ARM`].
enum OutputDisposition {
    /// The `none` arm — always silent and free, from every event. Nothing committed,
    /// nothing logged, nothing traced.
    Ignore,
    /// The single non-`none` arm this event honors. The caller commits its payload
    /// exactly as before this table existed.
    Honored,
    /// A non-`none` arm this event does not honor — a loud, non-fatal fault. Carries
    /// the arm name for the log line and trace record.
    Fault(&'static str),
}

/// Classify `output` for `event` at `phase` against that phase's honored-arm table.
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),
    }
}

/// The one-line diagnostic written to `logs/hook-<name>.log` (and mirrored in the
/// `hook_dispatch_error` trace record's fields) when a hook returns a non-`none`
/// arm the event does not honor. Names the hook, the event's WIT function name, and
/// the discarded arm so the fault is diagnosable without reading any Rust source.
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"
    )
}

/// Something went wrong dispatching one event to one hook, in a way the session survives.
/// Buffered by [`HookRuntime`] and drained via [`HookRuntime::drain_dispatch_faults`] so the
/// agent loop can write it to `trace.jsonl` as a `hook_dispatch_error` event.
///
/// Raised for a blocking hook's unsupported-arm return, and for all four ways an async hook
/// can fail visibly: an unsupported arm, a genuine `Err`, a dropped event, and a drain
/// timeout. Only `on-stage` faults stay log-only — that path runs during staging, before the
/// trace writer exists.
#[derive(Debug, Clone)]
pub(crate) struct DispatchFault {
    /// Manifest name of the hook the fault is attributed to.
    pub hook_name: String,
    /// WIT lifecycle function name the fault arose from (e.g. `on-tool-call`), or
    /// [`FAULT_EVENT_DRAIN`] for a fault not tied to a single call.
    pub event: String,
    /// The unsupported `hook-output` arm name (e.g. `write-manifests`), or one of
    /// [`FAULT_ARM_ERROR`] / [`FAULT_ARM_QUEUE_OVERFLOW`] / [`FAULT_ARM_TIMEOUT`] for a fault
    /// with no arm to name.
    pub arm: String,
}

struct HookInstance {
    name: String,
    config: HookConfig,
    store: Store<HookStoreState>,
    /// Name-based dispatch table keyed by the WIT function name
    /// (e.g. `"on-session-start"`). Always holds every entry in [`REQUIRED_HOOK_FNS`];
    /// entries from [`OPTIONAL_HOOK_FNS`] are present only if the component exports them.
    funcs: HashMap<String, wasmtime::component::Func>,
}

/// One async hook's dispatch endpoint, held by [`HookRuntime`] for the life of the session.
///
/// The hook's [`HookInstance`] is not here: it was moved into [`Self::worker`] at
/// instantiation and stays there, so the only thing the agent loop ever does to an async hook
/// is put a job on [`Self::tx`]. That is what makes `execution_mode: async` mean "off the
/// critical path" — and, because the worker holds one instance for the whole session, what
/// lets a hook accumulate state across events the way a blocking hook can.
struct AsyncHookQueue {
    name: String,
    /// The hook's own declared binding, copied out of its `HookConfig` at instantiation so
    /// dispatch can filter events without reaching into the worker's instance.
    binding: HookBinding,
    /// The operator's declared full-queue policy for this hook.
    on_overflow: HookOverflowPolicy,
    /// Bounded FIFO of pending calls, [`ASYNC_HOOK_QUEUE_DEPTH`] deep. Dropping it is what
    /// tells the worker to finish what it has and return; see
    /// [`HookRuntime::drain_async_hooks`].
    tx: mpsc::Sender<AsyncHookJob>,
    /// The one task draining `tx`'s receiver, one call at a time and in order.
    worker: tokio::task::JoinHandle<()>,
    /// Events discarded because the queue was full under [`HookOverflowPolicy::Drop`].
    /// Reported per event as a [`FAULT_ARM_QUEUE_OVERFLOW`] fault and summarized in the
    /// hook's log at drain time.
    dropped: u64,
}

/// One queued lifecycle call for an async hook.
///
/// Carries the pieces of session state a hook call needs that only the `&mut self` dispatch
/// path can read, captured at *enqueue* time: a call that waits in the queue must still
/// report the elapsed time and totals as of the event, not as of whenever the worker reached
/// it.
struct AsyncHookJob {
    event: HookEvent,
    elapsed: Duration,
    totals: HookTotals,
}

struct HookStoreState {
    /// Resource limiter for this store, registered via `Store::limiter`, and the record of
    /// any growth request it denied — read back by `classify_guest_failure` in
    /// [`call_typed`] so a limit trap reads distinguishably in `logs/hook-<name>.log`.
    limits: ExecutionLimiter,
    table: ResourceTable,
    wasi: WasiCtx,
    /// wasi-http context. A hook's *only* route to the network: [`build_wasi_ctx`] grants no
    /// raw WASI socket capability, so `wasi:http/outgoing-handler` — filtered by
    /// `http_hooks` below — is the whole outbound surface, exactly as for capsules and tools.
    http: WasiHttpCtx,
    /// Per-hook allow-list enforcement, built from this hook's `HookCapabilityGrant`. An
    /// empty rule set (the default, for a hook the operator granted no `capabilities.network`)
    /// denies every request.
    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,
    /// Fires once per task, immediately before that task's agent loop runs.
    /// For `task_acceptance: none`/`single` this coincides with `SessionStart`;
    /// for `task_acceptance: queue` it fires once per queued task.
    TaskStart {
        task_id: String,
        context_id: String,
        source: String,
        input_bytes: u64,
        /// Token ceiling a `seed-context` returned from this event is enforced
        /// against: `floor(context.max_tokens * context.seed_budget)`. `0` when
        /// the capsule declares no `context.max_tokens`, which the WIT contract
        /// reads as "not computed" rather than "unbounded".
        budget_tokens: u64,
        /// The capsule's `context.max_tokens`, from
        /// [`crate::runtime::resolve_context_window`], or `0` when the manifest
        /// declares no `context:` block.
        context_window: u64,
        /// Tokens already occupied by the conversation record this task will load
        /// under `lifecycle.conversation: threaded`. `0` under every other
        /// conversation mode, and read the same way as `budget_tokens`.
        /// Without it a hook seeding into a threaded context duplicates history
        /// it cannot see.
        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,
        /// The exact tool input JSON the call received, untruncated.
        input: String,
        output_bytes: u64,
        duration_ms: u64,
        status: String,
    },
    Shell {
        turn: u32,
        /// The program that ran, canonicalized against the host `PATH` where possible
        /// (else the bare invoked name).
        binary: String,
        command: String,
        /// The exact argument list handed to the executable, untruncated — unlike `command`,
        /// which is clipped for display.
        argv: Vec<String>,
        /// The `-c` body for the interpreter form, `None` for every other form.
        script: Option<String>,
        /// The build-tool recipe body the call's own resolution read, `None` when it named
        /// none the runtime could resolve.
        recipe: Option<String>,
        exit_code: i32,
        stdout: String,
        stderr: String,
        stdout_bytes: u64,
        stderr_bytes: u64,
        duration_ms: u64,
    },
    /// Fires when the session token threshold is reached. Routed through the same
    /// shared dispatch path as every other event; a blocking hook may return
    /// `replace-context` to swap out the conversation history.
    Compaction {
        messages: Vec<Message>,
        session_tokens: u64,
        threshold: f64,
        /// Manifest-configured model for the hook's own summarization call, from
        /// `inference.compaction.model`. `None` when the manifest leaves it unset —
        /// resolving that to a concrete model is the receiving hook's job, not ours.
        model: Option<String>,
        /// Manifest-configured system-prompt override for that call, from
        /// `inference.compaction.system_prompt`. `None` when the manifest leaves it
        /// unset — picking a default prompt is the receiving hook's job, not ours.
        system_prompt: Option<String>,
    },
    /// Fires once per task, immediately after that task's agent loop returns.
    TaskEnd {
        task_id: String,
        exit_status: String,
    },
    SessionEnd {
        total_turns: u32,
        exit_status: String,
    },
}

/// Outcome of dispatching one event to one blocking hook. The shared dispatch loop
/// funnels every event through [`call_hook`], which returns one of these regardless
/// of event type; the caller (`dispatch`) collects the relevant variant per event.
enum HookCallResult {
    /// No committable output (hook returned `none`/`write-manifests`, produced an
    /// artifact for an event that doesn't forward artifacts, or the optional
    /// function is absent from this component).
    None,
    /// A structured artifact payload (only forwarded for `on-inference`).
    Artifact(HookArtifact),
    /// A replacement conversation history (only meaningful for `on-compaction`).
    ReplaceContext(Vec<Message>),
    /// Messages proposed as the head of the task's context (only honored for
    /// `on-task-start`). Carries the proposing hook's manifest name; the runtime
    /// enforces the budget and records the outcome against it. See [`HookSeed`].
    SeedContext {
        hook_name: String,
        messages: Vec<Message>,
    },
    /// A control decision to reopen the task (only honored for `on-task-end`).
    /// Carries the requesting hook's manifest name and its feedback `reason`; the
    /// agent loop re-runs with that feedback injected into the task content, up to
    /// the capsule's reopen/turn budget. See [`TaskReopen`].
    Reopen { hook_name: String, reason: String },
    /// The hook returned a non-`none` arm the event does not honor. Non-fatal: the
    /// caller logs it (and, for the shared dispatch path, buffers it for the trace)
    /// and continues as if the hook had returned `none`. Carries the event's WIT
    /// function name and the discarded arm name.
    UnsupportedArm { event: String, arm: String },
}

/// What one pass of [`HookRuntime::dispatch`] produced, across every bound blocking hook.
///
/// One struct rather than a tuple because each field belongs to a different event and only
/// one caller ever reads each: a positional tuple makes every new honored arm rewrite every
/// call site, and makes it possible to read the wrong half without the compiler noticing.
struct DispatchOutcome {
    /// Every `on-inference` `HookOutput::Artifact`, in hook-registration order.
    artifacts: Vec<HookArtifact>,
    /// The first `on-compaction` `HookOutput::ReplaceContext`.
    replacement: Option<Vec<Message>>,
    /// The first `on-task-start` `HookOutput::SeedContext`.
    seed: Option<HookSeed>,
    /// The message of the first bound hook that returned `Err` for this event. Only
    /// `on-compaction` treats it as fatal.
    first_error: Option<String>,
    /// The first `on-task-end` `HookOutput::ReopenTask`.
    reopen: Option<TaskReopen>,
}

/// A blocking `on-task-end` hook returned `reopen-task(reason)`: a control
/// decision that the task's agent loop should re-run with `reason` injected as
/// feedback rather than being finalized. Surfaced by [`HookRuntime::dispatch_task_end`]
/// and acted on by the runtime's per-task reopen loop, subject to the capsule's
/// `lifecycle.max_task_reopens` budget and `inference.max_turns` ceiling.
#[derive(Debug, Clone)]
pub(crate) struct TaskReopen {
    /// Manifest name of the hook that requested the reopen.
    pub hook_name: String,
    /// Feedback text the hook wants injected into the reopened task's content.
    pub reason: String,
}

/// A blocking `on-task-start` hook returned `seed-context(messages)`: context it proposes
/// as the head of this task's message list, ahead of any loaded history and the task
/// message. Surfaced by [`HookRuntime::dispatch_task_start`]; what the agent loop does with
/// it depends on how it measures against the task's seed budget, which is the hook's
/// proposal and the runtime's decision respectively.
///
/// The messages are chronological, oldest first, matching the loop's own `messages` — that
/// is what makes "drop from the front" a well-defined trim.
#[derive(Debug, Clone)]
pub(crate) struct HookSeed {
    /// Manifest name of the hook that proposed the seed, recorded on the `context_seed`
    /// trace event and on any rejection fault.
    pub hook_name: String,
    /// The proposed messages, exactly as the hook returned them.
    pub messages: Vec<Message>,
}

/// An artifact emitted by a blocking hook via `HookOutput::Artifact`.
/// Carries the hook's manifest name so callers can identify the source without
/// any hardcoded strings.
#[derive(Debug, Clone)]
pub(crate) struct HookArtifact {
    /// Manifest name of the hook that emitted this artifact.
    pub hook_name: String,
    /// JSON payload string returned by the hook.
    pub payload: String,
}

#[derive(Debug, Clone)]
pub(crate) struct ShellDispatchInfo {
    /// The program that was actually invoked — a canonical absolute path when the host
    /// `PATH` resolves the invoked name, else the bare name. Distinct from `command`,
    /// which carries the argument list alone and therefore never names the binary.
    pub binary: String,
    pub command: String,
    /// The argument list `execute_shell` was handed, as resolved by
    /// `runtime::resolve_shell_call` — the same value a policy hook was shown before the call.
    pub argv: Vec<String>,
    /// The `-c` body for the interpreter form, `None` otherwise.
    pub script: Option<String>,
    /// The build-tool recipe body the same resolution read, so the observation event carries
    /// what the decision point was shown.
    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,
    /// The `capabilities.resources` field this call was killed for exceeding, when the evidence
    /// names exactly one. Carried alongside `exit_code` rather than folded into it because a
    /// signal-killed process has no exit code to fold it into.
    pub resource_limit: Option<String>,
}

/// What is about to run, resolved by the runtime from the tool name and its input. The hook
/// holds no filesystem grant and cannot resolve any of this itself.
///
/// Built by `CapsuleStoreState::resolve_call` from the same resolver the spawn path uses, so
/// the argv a policy hook decides on is the argv the executable receives.
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 {
    /// The WIT lifecycle function whose decision point gates this call.
    pub(crate) fn event_name(&self) -> &'static str {
        match self {
            Self::Shell { .. } => "on-shell",
            Self::Tool { .. } => "on-tool-call",
        }
    }

    /// What a denial names as the thing refused: the resolved executable for a shell call, the
    /// tool name for everything else.
    pub(crate) fn target(&self) -> &str {
        match self {
            Self::Shell { binary, .. } => binary,
            Self::Tool { tool_name, .. } => tool_name,
        }
    }

    /// The binding a hook must declare to be asked about this call.
    fn binding(&self) -> HookBinding {
        match self {
            Self::Shell { .. } => HookBinding::OnShell,
            Self::Tool { .. } => HookBinding::OnToolCall,
        }
    }
}

/// The outcome of the decision-point dispatch. Two variants and no more: there is no arm that
/// permits, and `Denied` carries no capability, path, grant or policy value. A hook can only
/// subtract from what the manifest already granted, and this type is where that is made
/// unrepresentable rather than merely intended.
pub(crate) enum CallDecision {
    Proceed,
    Denied { hook_name: String, reason: String },
}

/// Env vars injected into every hook's WASI context. Group avoids a >7-arg constructor.
#[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>,
    /// The session's [`crate::formation::FormationId`], injected as
    /// [`crate::formation::FORMATION_ID_ENV`]. Taken from the session and never from the process
    /// environment, so two sessions staged in one process each hand their hooks their own.
    pub formation_id: Option<&'a str>,
    /// The session's formation membership. Its callees, at their virtual addresses, are injected
    /// as [`crate::formation::FORMATION_PEERS_ENV`] right after the formation id, and the hook's
    /// egress reaches them through it. `Some` only for a formation member; taken from the
    /// session, never from the process.
    pub formation_member: Option<&'a Arc<crate::formation_credentials::FormationMember>>,
}

/// Dispatch `on-stage` for all blocking hooks with a matching binding.
///
/// Called from `stage_session` which may be in a sync context; this function is sync
/// and uses a temporary current-thread runtime internally.
#[allow(clippy::too_many_arguments)]
pub(crate) fn dispatch_stage(
    engine: &wasmtime::Engine,
    workdir: &WorkdirRoot,
    records_root: &WorkdirRoot,
    runtime_root: &WorkdirRoot,
    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,
                records_root,
                staged,
                &evt,
                env_vars,
                limits,
            )
            .await
            {
                // `on-stage` runs during staging, before `trace.jsonl` exists, so an
                // unsupported-arm fault here is logged but never traced — unlike the
                // shared `dispatch` path. The honored arm is decided by the same
                // `HONORED_OUTPUT_ARM` table via `classify_output`.
                Ok(output) => match classify_output(DispatchPhase::Observe, "on-stage", &output) {
                    OutputDisposition::Honored => {
                        // `on-stage` honors only `write-manifests`.
                        if let HookOutput::WriteManifests(manifests) = output {
                            for m in manifests {
                                // A refusal here is on the latch `stage_session` drains, which
                                // fails the launch.
                                if let Err(e) = runtime_root.write_file(
                                    format!("tools/{}/{PACKED_MANIFEST_ENTRY}", m.binary_name),
                                    m.content.as_bytes(),
                                    DEFAULT_FILE_MODE,
                                ) {
                                    log_hook_error(
                                        records_root,
                                        &staged.name,
                                        &format!(
                                            "failed to write manifest for {}: {e}",
                                            m.binary_name
                                        ),
                                    )
                                    .await;
                                }
                            }
                        }
                    }
                    OutputDisposition::Fault(arm) => {
                        log_hook_error(
                            records_root,
                            &staged.name,
                            &format_dispatch_fault(&staged.name, "on-stage", arm),
                        )
                        .await;
                    }
                    OutputDisposition::Ignore => {}
                },
                Err(err) => log_hook_error(records_root, &staged.name, &err).await,
            }
        }
        Ok(())
    })
}

async fn call_stage_once(
    engine: &wasmtime::Engine,
    workdir: &WorkdirRoot,
    records_root: &WorkdirRoot,
    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())?;
    // `on-stage` runs during staging, long before an inference driver exists —
    // the import is defined so an inference-importing hook still links, and
    // always errors.
    add_inference_to_linker(
        &mut linker,
        format!("hook:{}", staged.name),
        Err(InferenceUnavailable::NotConfigured),
    )?;
    // `on-stage` runs before any task exists. A granted hook gets a state of its own so its
    // reads truthfully report `no-task` rather than `not-granted`; the throwaway store this
    // instance lives on is discarded with it.
    add_task_io_to_linker(
        &mut linker,
        staged
            .grant
            .task_io_read
            .then(|| Arc::new(TaskIoState::new())),
    )?;
    // `on-stage` runs before any task, so a granted hook gets a state with no context in scope:
    // its reads truthfully report an empty page rather than `not-granted`.
    add_conversation_to_linker(
        &mut linker,
        staged
            .grant
            .conversation_read
            .then(|| Arc::new(ConversationState::new(None, records_root))),
    )?;
    // Counting text reaches nothing, so the counter is defined on every linker
    // with no grant to consult.
    add_tokens_to_linker(&mut linker)?;

    let state = HookStoreState {
        limits: limits.limiter(),
        table: ResourceTable::new(),
        wasi: build_wasi_ctx(
            workdir.path(),
            env_vars,
            &staged.grant,
            staged.gateway.as_deref(),
        )
        .map_err(|e| e.to_string())?,
        http: WasiHttpCtx::new(),
        http_hooks: NetworkPolicyHooks {
            network_allow_rules: staged.grant.network_allow_rules.clone(),
            gateway: staged.gateway.clone(),
            formation: env_vars.formation_member.cloned(),
            // A hook runs for no task of its own, so a formation call from it stamps
            // `untrusted`.
            task_provenance: 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))?;

    // `on-stage` runs on its own throwaway store, so this cannot reuse the
    // blocking-hook path.
    call_stage_lift::<HookOutput>(&mut store, &func, evt, &staged.name, limits).await
}

/// Call `on-stage` on a throwaway store, lifting its `result<O, string>` where `O` is
/// the version-appropriate `hook-output` type. Returns the hook's own returned error
/// as the `Err` string, exactly as the pre-versioning inline call did.
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())?;
    // Fresh budget for the call itself, so instantiation cost cannot eat into it.
    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 {
    /// Instantiate every hook component — blocking and async alike — exactly once, and start
    /// one worker task per async hook to drain that hook's queue.
    ///
    /// Both modes go through the same [`instantiate_hook`] call with the same `env_vars`, so
    /// an async hook gets the session's real environment (`MURMUR_OTEL_ENDPOINT` and friends)
    /// and keeps one `Store` — one linear memory, one set of guest-side globals — for the
    /// whole session. Only what happens *after* instantiation differs: a blocking hook's
    /// instance is called inline by `dispatch`, an async hook's is owned by its worker.
    ///
    /// `instantiate_async` is required because the engine has `async_support(true)`.
    ///
    /// Must be called from inside a `tokio::task::LocalSet` whenever any staged hook is
    /// async: the workers are `spawn_local` tasks, tied to the thread running the agent loop.
    /// The session's `LocalSet` in `runtime.rs` is that context.
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn new(
        engine: &wasmtime::Engine,
        records_root: &WorkdirRoot,
        accessible_workdir: &Path,
        staged_hooks: Vec<StagedHookArtifact>,
        context: SessionContextData,
        env_vars: HookEnvVars<'_>,
        limits: ExecutionLimits,
        inference: Result<Arc<HookInferenceCtx>, InferenceUnavailable>,
        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, records_root)),
        };

        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(),
                        records_root.clone(),
                        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 {
            records_root: records_root.clone(),
            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,
        })
    }

    /// Whole-launch turn count, accumulated across every task processed by this
    /// runtime (one increment per `Inference` event). Read by `runtime.rs` when it
    /// fires the once-per-launch `SessionEnd` event so `total-turns` is a launch
    /// aggregate rather than a single task's count.
    pub(crate) fn total_turns(&self) -> u32 {
        self.total_turns
    }

    /// Take every `run-inference` trace record buffered since the last drain.
    /// The agent loop calls this after hook dispatch and writes each one through
    /// the session's `TraceWriter`/`OtelEmitter` — one record per call, success
    /// or failure, tagged `hook:<name>`.
    pub(crate) fn drain_inference_records(&self) -> Vec<HookInferenceRecord> {
        self.inference
            .as_ref()
            .map(|ctx| ctx.drain_records())
            .unwrap_or_default()
    }

    /// Take every `run-inference` call a spend ceiling refused since the last drain. Drained by
    /// the agent loop beside [`Self::drain_inference_records`].
    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()
    }

    /// Put a task in scope for one attempt of the agent loop, with the two input forms the
    /// `murmur:task-io/read` import distinguishes: `original` is the task before any reopen
    /// feedback was appended, `as_given` is exactly what this attempt's loop is handed.
    /// Clears any previous attempt's recorded output.
    pub(crate) fn begin_task_attempt(&self, original: String, as_given: String) {
        self.task_io.begin_task_attempt(original, as_given);
    }

    /// Record the result text the agent loop produced for the attempt in scope. Called from
    /// each transport's single result-write funnel, so every terminal path that produces
    /// result text records it and no terminal path can produce one the import cannot see.
    pub(crate) fn record_task_output(&self, text: &str) {
        self.task_io.record_output(text);
    }

    /// Take the task out of scope. Every `murmur:task-io/read` call after this reports
    /// `no-task` until the next [`Self::begin_task_attempt`].
    pub(crate) fn end_task(&self) {
        self.task_io.end_task();
    }

    /// Take every dispatch fault produced since the last drain, from both sources: those
    /// buffered inline by [`Self::dispatch`] and those sent over [`Self::fault_tx`] by an
    /// async hook's worker. The agent loop calls this immediately before each `session_end`
    /// write and records each one through the session's `TraceWriter` as a
    /// `hook_dispatch_error` event, so no fault is lost regardless of which exit path the
    /// session takes. `on-stage` faults are logged only and never appear here.
    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
    }

    /// Finish every async hook's outstanding work, under one shared bounded budget.
    ///
    /// Closing a hook's queue is what ends its worker: the worker completes the jobs already
    /// enqueued, in order, sees the channel close, and returns. This awaits that — which is
    /// the whole point, because the session's `LocalSet` is torn down immediately afterwards
    /// and would otherwise drop a worker mid-call, taking a queued `on-session-end` export
    /// with it.
    ///
    /// Bounded by [`ASYNC_HOOK_DRAIN_TIMEOUT`] across all hooks together, so N slow hooks
    /// cannot multiply the delay. A hook that runs out of budget is abandoned, logged, and
    /// recorded as a [`FAULT_ARM_TIMEOUT`] fault for `trace.jsonl` — call this *before*
    /// `drain_dispatch_faults` so those faults are still picked up.
    ///
    /// Idempotent: it takes the queues, so a second call has nothing left to drain. Async
    /// hooks dispatched to after it are silently skipped, which is correct at session end and
    /// is the only place it is called.
    pub(crate) async fn drain_async_hooks(&mut self) {
        self.drain_async_hooks_within(ASYNC_HOOK_DRAIN_TIMEOUT)
            .await;
    }

    /// [`Self::drain_async_hooks`] with an explicit budget, so the timeout path can be
    /// exercised without waiting out the production one.
    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.records_root, &name, &summary).await;
            }

            let remaining = deadline.saturating_duration_since(Instant::now());
            match tokio::time::timeout(remaining, &mut worker).await {
                // A worker only returns after its queue is empty, so this is the drained case.
                Ok(Ok(())) => {}
                Ok(Err(join_error)) => {
                    let message = format!("hook '{name}' worker task failed: {join_error}");
                    log_hook_error(&self.records_root, &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.records_root, &name, &message).await;
                    self.dispatch_faults.push(DispatchFault {
                        hook_name: name,
                        event: FAULT_EVENT_DRAIN.to_string(),
                        arm: FAULT_ARM_TIMEOUT.to_string(),
                    });
                }
            }
        }
    }

    /// Dispatch `event` to all bound hooks. Returns every artifact emitted by
    /// any blocking hook via `HookOutput::Artifact`, in hook-registration order.
    /// An empty `Vec` means no hook produced an artifact for this event.
    pub(crate) async fn emit(
        &mut self,
        workdir: &WorkdirRoot,
        event: HookEvent,
    ) -> Vec<HookArtifact> {
        self.dispatch(workdir, event).await.artifacts
    }

    /// Whether this session has a policy hook bound at all — a blocking hook declaring
    /// `commit_policy: deny` on `on-shell` or `on-tool-call`.
    ///
    /// A single boolean read, computed once at [`Self::new`]. `false` is the whole of the
    /// non-gated path: the agent loop resolves nothing, dispatches nothing and allocates
    /// nothing extra before a tool call.
    pub(crate) fn gates_calls(&self) -> bool {
        self.gates_calls
    }

    /// Ask every bound policy hook whether `call` may run, and stop at the first refusal.
    ///
    /// The decision point. It sits inside the already-granted path — after the manifest
    /// capability check and immediately before the call is dispatched — and its only possible
    /// effect on the runtime is to *skip* work. It takes no capability policy and returns no
    /// grant: [`CallDecision`] has no arm that permits, so nothing here can widen what the
    /// manifest allowed.
    ///
    /// **Fail-closed, and not configurable.** The call proceeds only on a clean
    /// `hook-output::none`. Every other outcome refuses it:
    ///
    /// | hook outcome | result |
    /// | --- | --- |
    /// | `Ok(none)` | proceed |
    /// | `Ok(deny(reason))`, `reason` non-blank | deny, with the hook's reason verbatim |
    /// | `Ok(deny(reason))`, `reason` blank | deny, with a runtime reason naming the defect |
    /// | `Ok(<any other arm>)` | deny, and a [`DispatchFault`] is recorded for that arm |
    /// | `Err(msg)` — trap, panic, epoch deadline, memory-limit kill, lift failure | deny, carrying `msg` |
    /// | function absent from the component | deny |
    ///
    /// There is no manifest field, environment variable, CLI flag or config key that changes
    /// any row of it, and that absence is deliberate: the inversion of the "hook errors are
    /// non-fatal" default is the whole point of a policy hook, and an escape hatch would make
    /// a broken hook silently permissive. The observation dispatch of these same two events
    /// keeps the non-fatal default, unchanged.
    ///
    /// The last row cannot be reached in practice — both gated functions are in
    /// [`REQUIRED_HOOK_FNS`], so a component missing either fails to instantiate — and is
    /// written as a refusal rather than left to that fact.
    pub(crate) async fn decide(
        &mut self,
        workdir: &WorkdirRoot,
        turn: u32,
        call: &ResolvedCall,
    ) -> CallDecision {
        let event_fn = call.event_name();
        let binding = call.binding();
        let mut decision = CallDecision::Proceed;
        // Buffered and applied after the loop: it holds `&mut self.blocking_hooks`, so nothing
        // else on `self` can be touched while it runs.
        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(),
                        // `none` is what tells the hook this call has not run, and is the only
                        // dispatch at which its `deny` is honored.
                        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,
                        // A refusal the agent cannot act on is still a refusal; the runtime
                        // supplies the reason the hook did not.
                        _ => 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"
                        )
                    }
                },
                // Unreachable: `on-shell` and `on-tool-call` are both required exports.
                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"
                ),
                // `classify_guest_failure` has already made `error` name the limit that fired,
                // for a trap, an epoch-deadline expiry or a memory-limit kill alike.
                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
    }

    /// Buffer a fault the runtime itself raised about a hook, for the same drain the
    /// dispatch-path faults go through.
    ///
    /// Dispatch reports what a hook *returned*; this reports what the runtime *did* with a
    /// return it honored — the only current caller is the seed commit refusing an
    /// over-budget or unbudgeted `seed-context` under [`FAULT_ARM_SEED_REJECTED`]. Both end
    /// up as `hook_dispatch_error` lines, because from a trace reader's side there is one
    /// question: which hook output did not take effect, and on which event.
    pub(crate) fn record_dispatch_fault(&mut self, fault: DispatchFault) {
        self.dispatch_faults.push(fault);
    }

    /// Shared dispatch path used by every Lifecycle Event. Iterates the blocking
    /// hooks (binding-filtered) and calls each inline, then enqueues the event on each
    /// matching async hook's queue and returns without waiting for any of them.
    ///
    /// Every caller reads only the [`DispatchOutcome`] field its event honors and ignores
    /// the rest; a per-hook error is still logged and the loop still continues regardless
    /// of which field the caller reads. Event-keyed side effects (the running
    /// token/tool-call/shell-call totals) run here so all events funnel through one place.
    async fn dispatch(&mut self, workdir: &WorkdirRoot, 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;
        // Faults are collected here and appended to `self.dispatch_faults` after the
        // loop: the loop holds `&mut self.blocking_hooks`, so `self` cannot be touched
        // otherwise while it runs.
        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,
                }) => {
                    // First seeding hook wins, mirroring `replacement`.
                    if seed.is_none() {
                        seed = Some(HookSeed {
                            hook_name,
                            messages,
                        });
                    }
                }
                Ok(HookCallResult::Reopen { hook_name, reason }) => {
                    // First reopen-requesting hook wins, mirroring `replacement`.
                    if reopen.is_none() {
                        reopen = Some(TaskReopen { hook_name, reason });
                    }
                }
                Ok(HookCallResult::None) => {}
                Ok(HookCallResult::UnsupportedArm { event: ev, arm }) => {
                    // Non-fatal: log it, buffer it for the trace, and continue as if the
                    // hook had returned `none`. It is not an `Err`, so it never becomes
                    // `first_error` — compaction must not fail the session over it.
                    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);

        // Async hooks: enqueue and move on. The call itself runs later on that hook's own
        // worker, against the instance it has been reusing since `new` — nothing here waits
        // for a hook to answer, which is what `execution_mode: async` buys.
        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 {
                // Never blocks: a full queue means the hook cannot keep up, and the operator
                // asked for the agent loop to stay moving. The discard is counted and traced.
                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(),
                        });
                    }
                    // Only reachable after `drain_async_hooks` has closed the queue, or if the
                    // worker died — both already reported at their own site.
                    Err(mpsc::error::TrySendError::Closed(_)) => {}
                },
                // The operator chose losslessness over latency: wait for the worker to make
                // room, putting this hook back on the critical path for as long as it is full.
                HookOverflowPolicy::Block => {
                    let _ = queue.tx.send(job).await;
                }
            }
        }

        DispatchOutcome {
            artifacts,
            replacement,
            seed,
            first_error,
            reopen,
        }
    }

    /// Fire `on-compaction` on all hooks with a matching binding.
    ///
    /// The three outcomes a caller must be able to tell apart:
    /// - `Ok(Some(messages))` — a blocking hook returned `replace-context`.
    /// - `Ok(None)` — no compaction-bound hook was invoked at all (or every one of
    ///   them returned `none`): the caller continues, uncompacted.
    /// - `Err(message)` — a compaction-bound hook *ran* and returned `Err`. There is
    ///   no safety net behind a declared compaction hook, so the caller must fail the
    ///   session rather than limp on with an over-budget context.
    ///
    /// A thin wrapper over the shared [`Self::dispatch`] path: it builds a
    /// `HookEvent::Compaction` and reads both the `replacement` and `first_error`
    /// halves of the result. An error wins over a replacement produced by some other
    /// hook in the same dispatch — a partially-failed compaction is still a failure.
    /// An async hook is only handed the event; its output is always discarded.
    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.records_root.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),
        }
    }

    /// Fire `on-task-start` on all hooks with a matching binding and surface the first
    /// [`HookSeed`] any blocking hook proposed via `seed-context`, or `None` if none did.
    ///
    /// A thin wrapper over the shared [`Self::dispatch`] path, mirroring
    /// [`Self::dispatch_task_end`]. First hook to return a seed wins, matching compaction:
    /// merging two hooks' proposals would need an ordering rule neither hook declared.
    /// A hook-returned `Err` is logged per-hook inside `dispatch` and never fails the task —
    /// a capsule whose memory hook broke must still run.
    #[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.records_root.clone();
        // The task's conversation becomes the one `murmur:conversation/read` answers about, from
        // this dispatch onward: a memory hook bound to `on-task-start` reads the record it is
        // about to seed from.
        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
    }

    /// Fire `on-task-end` on all hooks with a matching binding and surface the first
    /// [`TaskReopen`] any blocking hook requested via `reopen-task`, or `None` if none
    /// did. A thin wrapper over the shared [`Self::dispatch`] path, mirroring
    /// [`Self::dispatch_compaction`]: it builds a `HookEvent::TaskEnd` and reads only
    /// the `reopen` field of the outcome. `on-task-end` honors no artifact or
    /// replacement arm, so those fields are discarded; async hooks are still only
    /// enqueued. A hook-returned `Err` is logged per-hook inside `dispatch`
    /// and does not abort the task, matching `emit`.
    pub(crate) async fn dispatch_task_end(
        &mut self,
        task_id: String,
        exit_status: String,
    ) -> Option<TaskReopen> {
        let workdir = self.records_root.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 {
        // `on-stage` is the only event never routed through `dispatch`, so `All`
        // (which excludes on-stage) matches every event that can reach here.
        HookBinding::All => true,
        HookBinding::OnStage => false, // on-stage fired separately during staging
        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 { .. }),
    }
}

/// The host state every hook linker is built against, whatever the grant on the entry: one copy
/// per launch, shared with each granted hook through an `Arc` and withheld from the rest.
struct HookHostState {
    task_io: Arc<TaskIoState>,
    conversation: Arc<ConversationState>,
}

/// Instantiate one staged hook against its own `Store`, resolve its lifecycle exports, and
/// hand back the reusable instance.
///
/// The single instantiation path for both execution modes — a hook's store, WASI context,
/// capability grant and dispatch table are built the same way whether the runtime will call
/// it inline or from a worker task. Called once per hook, from [`HookRuntime::new`].
async fn instantiate_hook(
    engine: &wasmtime::Engine,
    project_dir: &Path,
    staged: &StagedHookArtifact,
    env_vars: &HookEnvVars<'_>,
    limits: ExecutionLimits,
    inference: Result<Arc<HookInferenceCtx>, InferenceUnavailable>,
    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)?;
    // An ungranted hook gets the import defined but backed by nothing, so it links exactly as
    // a granted one does and every read returns `not-granted`.
    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,
            staged.gateway.as_deref(),
        )?,
        http: WasiHttpCtx::new(),
        http_hooks: NetworkPolicyHooks {
            network_allow_rules: staged.grant.network_allow_rules.clone(),
            gateway: staged.gateway.clone(),
            formation: env_vars.formation_member.cloned(),
            // A hook runs for no task of its own, so a formation call from it stamps
            // `untrusted`.
            task_provenance: 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,
    })
}

/// Build the name-based dispatch table for a hook component: every function in
/// [`REQUIRED_HOOK_FNS`] must resolve (a miss returns `Err(missing(fn_name))`),
/// while [`OPTIONAL_HOOK_FNS`] are inserted only when the component exports them —
/// this is what lets a pre-`on-task-start`/`on-task-end` component instantiate
/// cleanly and simply never receive those two events.
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,
        }
    }
}

/// Routes every hook outbound HTTP request through the same `NetworkPolicyHooks` gate that
/// `CapsuleStoreState`/`ToolStoreState` use — one enforcement implementation, three guest
/// kinds. The rules come from the hook's own grant, so a hook is never widened by the
/// capsule-wide `capabilities.network.allow`.
impl WasiHttpView for HookStoreState {
    fn http(&mut self) -> WasiHttpCtxView<'_> {
        WasiHttpCtxView {
            ctx: &mut self.http,
            table: &mut self.table,
            hooks: &mut self.http_hooks,
        }
    }
}

/// Invoke one named `murmur:hook/lifecycle` function on a hook instance with a single
/// record argument, awaiting its `result<hook-output, string>`.
///
/// Returns `Ok(None)` when the function is absent from this component's dispatch table —
/// only ever true for an [`OPTIONAL_HOOK_FNS`] entry, which is how a hook that predates
/// `on-task-start`/`on-task-end` is silently skipped for those events. A hook-returned
/// `Err` is propagated as the call error, matching the pre-refactor behavior.
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))
}

/// Invoke one component `func` with a single record arg and lift its
/// `result<hook-output, string>`. The `Err` variant carries either an infra
/// failure or the hook's own returned error string.
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())?;

    // A blocking hook's store outlives every event it handles, so each lifecycle call is
    // given its own fresh budget rather than sharing one across the whole session.
    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) => {
            // Still a plain `String` on the same `Err` path an ordinary hook error takes —
            // `dispatch` isolates it via `log_hook_error` and the session continues.
            // Classification only makes the message name the limit that fired.
            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
}

/// Call one hook instance for a runtime event via its name-based dispatch table.
///
/// Maps each `HookEvent` to its WIT function name, builds the matching record, and
/// invokes it through [`call_typed`]. Only `on-inference` forwards a `HookOutput::Artifact`
/// and only `on-compaction` forwards a `HookOutput::ReplaceContext`; every other event
/// discards the output. A missing optional function yields `HookCallResult::None`.
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(),
                // `some` is what makes this the observation dispatch: the call has run, and
                // nothing the hook returns can unmake it.
                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,
        } => {
            // `binary`, `argv` and `script` are not truncated the way `command` is — a clipped
            // path names a different file, or none, and a clipped argv names a different call.
            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?
        }
    };

    // A single decision against `HONORED_OUTPUT_ARM`: commit the one arm this event
    // honors, stay silent for `none` and for an absent optional function, and turn any
    // other non-`none` arm into an unsupported-arm fault the caller logs and traces.
    let event_fn = event_fn_name(event);
    Ok(match output {
        // Optional function absent from this component — never a fault; the hook
        // simply is not dispatched for this event.
        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),
                // `seed-context` is honored only by `on-task-start`; the runtime's
                // seed-application path reads this off `dispatch_task_start`.
                HookOutput::SeedContext(messages) => HookCallResult::SeedContext {
                    hook_name: hook.name.clone(),
                    messages,
                },
                // `reopen-task` is honored only by `on-task-end`; the runtime's
                // reopen loop reads this off `dispatch_task_end`.
                HookOutput::ReopenTask(reason) => HookCallResult::Reopen {
                    hook_name: hook.name.clone(),
                    reason,
                },
                // `write-manifests` is honored only by `on-stage`, which is never
                // dispatched through `call_hook`, so it cannot reach this arm as
                // `Honored`; `none` classifies as `Ignore` before it gets here.
                // `deny` is honored only at the decision point, which never routes through
                // `call_hook`; from here it has already classified as a fault.
                HookOutput::WriteManifests(_) | HookOutput::Deny(_) | HookOutput::None => {
                    HookCallResult::None
                }
            },
            OutputDisposition::Fault(arm) => HookCallResult::UnsupportedArm {
                event: event_fn.to_string(),
                arm: arm.to_string(),
            },
        },
    })
}

/// The WIT lifecycle function name a [`HookEvent`] dispatches to. Used to key
/// [`HONORED_OUTPUT_ARM`] and to name the event in an unsupported-arm fault.
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",
    }
}

/// The body of one async hook's worker task: drain `jobs` forever, one call at a time.
///
/// Serialization is the point, not an implementation detail — `recv().await` hands over one
/// job, the hook call runs to completion against the single reused `instance`, and only then
/// is the next job taken. So calls to a given async hook happen in dispatch order and never
/// overlap, which is what makes a `Store` (and any state the guest keeps in it) safe to reuse.
///
/// Returns when the queue is closed and empty, i.e. when [`HookRuntime::drain_async_hooks`]
/// drops the sender.
async fn run_async_hook_worker(
    mut instance: HookInstance,
    mut jobs: mpsc::Receiver<AsyncHookJob>,
    context: SessionContextData,
    workdir: WorkdirRoot,
    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,
                });
            }
            // Any committable output an async hook returns is discarded here, including an
            // arm this event would honor for a blocking hook: nothing waited for the answer,
            // so there is nowhere left to apply it. Permanent, not a gap.
            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(),
                });
            }
        }
    }
}

/// Build a WASI context for a hook instance, governed by `grant` — the capability block the
/// capsule operator declared on this hook's entry in their own manifest.
///
/// Default-deny on both axes, and the default is what an ungranted hook gets:
///
/// - **Network.** Nothing here grants network capability. `inherit_network()` and
///   `allow_ip_name_lookup(true)` are deliberately absent, so a hook has no raw WASI sockets
///   under any grant. Its only outbound route is `wasi:http/outgoing-handler`, linked at each
///   instantiation site and filtered by `NetworkPolicyHooks` against
///   `grant.network_allow_rules` — the same gated path capsules and tools use. An empty rule
///   list denies every request.
/// - **Filesystem.** No preopened directory unless `grant.filesystem_scope` names one, in
///   which case exactly one directory is preopened: `root_dir/<scope>`, mounted as the hook's
///   current directory `"."`. Nothing above or beside that subtree is reachable, because a
///   guest can only name paths under a preopen. The scope was already validated (relative,
///   non-escaping) by `HookCapabilityGrant::derive` at staging time; it is created here if it
///   does not exist, and a creation failure fails the instantiation rather than silently
///   downgrading to no access. A grant declaring `capabilities.filesystem.records: true` instead
///   preopens the session's records directory's `hooks/` as `"."`, the directory staging created
///   and named in `grant.records_hooks_dir`; one with no such directory fails the instantiation.
/// - **Configuration.** One environment variable, [`ARTIFACT_CONFIG_ENV`], added only when
///   `grant.config_json` carries one — the `config:` block this hook's own manifest entry
///   declared, lowered to compact JSON at staging. It rides the grant rather than
///   [`HookEnvVars`], which is session-wide, so one hook's config never reaches another's.
/// - **Durable state.** A second, independent preopen at [`STATE_PREOPEN_NAME`], added only when
///   `grant.state_dir` names one — the host store `capabilities.state` granted, already created at
///   `0700` by the staging path and outside every workdir. The two filesystem axes do not imply
///   each other in either direction: a hook may hold a durable store with no project directory at
///   all, and a project directory with no store.
///
/// `root_dir` is the hook's working directory: the session dir for `on-stage`, the project
/// directory for blocking and async hooks.
fn build_wasi_ctx(
    root_dir: &Path,
    env: &HookEnvVars<'_>,
    grant: &HookCapabilityGrant,
    gateway: Option<&crate::credential_gateway::CredentialGateway>,
) -> Result<WasiCtx, RuntimeError> {
    // No host env inheritance: the explicit injections below are the entire environment a hook
    // component ever sees. Hooks have no manifest-declared allowlist because no hook artifact
    // reads an arbitrary host var, so the only operator-authored value that can reach one is the
    // `config:` block on its own entry, which arrives here already lowered on the grant.
    let mut builder = WasiCtxBuilder::new();
    builder.inherit_stdio();

    // First, so the runtime-owned `MURMUR_*` variables below are applied after it and keep
    // precedence: a `config:` block can carry any key an operator writes, and none of them may
    // shadow a value the runtime owns.
    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);
    }
    // Only a hook whose own entry declares `gateway:` is told where the gateway is.
    if let Some(gateway) = gateway {
        let (name, value) = crate::runtime::gateway_env_pair(gateway);
        builder.env(name, value);
    }
    if let Some(id) = env.formation_id {
        builder.env(crate::formation::FORMATION_ID_ENV, id);
    }
    if let Some(peers) = env.formation_member.and_then(|member| member.guest_peers()) {
        let (name, value) = peers.env_pair();
        builder.env(name, value);
    }
    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, ".", FsPerms::ReadWrite)
            .map_err(|err| RuntimeError::wasi(scoped_dir, err.to_string()))?;
    } else if grant.filesystem_records {
        // Fail closed: a records grant with no staged directory fails the instantiation rather
        // than preopening anything else as ".".
        let hooks_dir = grant.records_hooks_dir.as_deref().ok_or_else(|| {
            RuntimeError::wasi(
                root_dir.to_path_buf(),
                "capabilities.filesystem.records is declared, but this session holds no \
                 records directory to preopen"
                    .to_string(),
            )
        })?;
        builder
            .preopened_dir(hooks_dir, ".", FsPerms::ReadWrite)
            .map_err(|err| RuntimeError::wasi(hooks_dir.to_path_buf(), err.to_string()))?;
    }

    if let Some(state_dir) = grant.state_dir.as_deref() {
        builder
            .preopened_dir(state_dir, STATE_PREOPEN_NAME, FsPerms::ReadWrite)
            .map_err(|err| RuntimeError::wasi(state_dir.to_path_buf(), err.to_string()))?;
    }

    Ok(builder.build())
}

/// Appends `error` to `logs/hook-<hook_name>.log` in the session's records directory. A refused
/// append is on `records_root`'s refusal latch; nothing else is reported.
///
/// Runs on the blocking pool so the `.await` yields: a checkpoint draining the latch can run
/// between an async hook worker's calls.
async fn log_hook_error(records_root: &WorkdirRoot, hook_name: &str, error: &str) {
    use std::io::Write;
    let records_root = records_root.clone();
    let hook_name = hook_name.to_string();
    let line = format!("{error}\n");
    let _ = tokio::task::spawn_blocking(move || {
        if let Ok(mut file) =
            records_root.open_append(format!("logs/hook-{hook_name}.log"), DEFAULT_FILE_MODE)
        {
            let _ = file.write_all(line.as_bytes());
        }
    })
    .await;
}

/// Hook runtimes other modules' tests drive the compaction path through.
#[cfg(test)]
pub(crate) mod test_support {
    use std::path::Path;

    use murmur_artifact::HookBinding;

    use super::{tests, HookRuntime};

    /// A runtime whose one `on-compaction` hook replaces the context with a single message:
    /// role `role`, content the `inference.compaction.model` it was handed.
    pub(crate) async fn echo_compaction_hooks(
        engine: &wasmtime::Engine,
        workdir: &Path,
        role: &str,
    ) -> HookRuntime {
        let double = tests::hook_compaction_echo_double(engine, "model", role);
        tests::new_with_hooks(
            engine,
            workdir,
            workdir,
            vec![tests::staged_double_named(
                "compactor",
                HookBinding::OnCompaction,
                double,
            )],
        )
        .await
        .expect("compaction double instantiates")
    }

    /// A runtime whose one `on-inference` hook, `caller`, calls `run-inference` through `ctx`
    /// on every agent-loop turn.
    pub(crate) async fn inference_calling_hooks(
        engine: &wasmtime::Engine,
        workdir: &Path,
        ctx: std::sync::Arc<crate::inference_import::HookInferenceCtx>,
    ) -> HookRuntime {
        HookRuntime::new(
            engine,
            &crate::workdir_fs::WorkdirRoot::open(workdir).unwrap(),
            workdir,
            vec![tests::staged_double_named(
                "caller",
                HookBinding::OnInference,
                tests::hook_inference_caller_double(engine),
            )],
            super::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(),
            },
            super::HookEnvVars::default(),
            crate::limits::ExecutionLimits::default(),
            Ok(ctx),
            None,
        )
        .await
        .expect("a hook importing murmur:runtime/inference links")
    }

    /// A runtime with no hook bound, so a compaction is declined for want of a replacement.
    pub(crate) async fn no_hooks(engine: &wasmtime::Engine, workdir: &Path) -> HookRuntime {
        tests::new_with_hooks(engine, workdir, workdir, Vec::new())
            .await
            .expect("empty hook set")
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::inference_import::INFERENCE_IFACE_VERSIONED;
    use tempfile::TempDir;
    use wasmtime::component::Component;

    /// A retired lifecycle instance name. The host accepts [`LIFECYCLE_IFACE`] and
    /// nothing else, so a double exporting this must fail to resolve — that is what
    /// the no-fallback tests below assert.
    const RETIRED_IFACE: &str = "murmur:hook/lifecycle@0.8.0";

    /// Engine configured like the production one (component model + async + epoch
    /// interruption) so a hand-authored WAT component double can be compiled and
    /// instantiated. Epoch interruption only arms the mechanism — a deadline fires solely
    /// when an [`EpochTicker`] is also running, which only the deadline test spawns.
    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")
    }

    /// A minimal `murmur:hook/lifecycle` component double exporting exactly `fn_names`,
    /// each backed by a trivial `func() -> ()` core function. Resolution in
    /// `resolve_hook_fns` only looks exports up by name — it never inspects signatures
    /// or calls them — so these stubs are sufficient to exercise the required-vs-optional
    /// instantiation logic (invariants 5 and 6).
    fn hook_double(engine: &wasmtime::Engine, fn_names: &[&str]) -> Component {
        // Default double exports the current versioned instance name, so the
        // required/optional suite exercises real resolution against the version the
        // host probes first.
        hook_double_iface(engine, LIFECYCLE_IFACE, fn_names)
    }

    /// Like [`hook_double`] but the exported lifecycle instance carries the given
    /// instance name, so tests can build a component that exports the versioned
    /// (`murmur:hook/lifecycle@0.9.0`), the legacy unversioned, or a
    /// deliberately-unmatched name to exercise `resolve_lifecycle_iface` and its
    /// hard-error path.
    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")
    }

    /// Default-deny staging, matching a manifest hook entry with no `capabilities:` block.
    fn staged_double(component: Component) -> StagedHookArtifact {
        staged_double_granted(component, HookCapabilityGrant::default())
    }

    /// [`staged_double`] with an explicit grant, for the capability suite.
    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(),
            gateway: None,
        }
    }

    pub(super) 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
    }

    /// [`new_with_hooks`] with an explicit limits block, so the deadline suite can ask for
    /// a budget short enough to observe inside a test.
    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
    }

    /// [`new_with_hooks_limited`] with an explicit environment, so the env-delivery suite can
    /// ask for the non-default values a real session passes in.
    async fn new_with_hooks_full(
        engine: &wasmtime::Engine,
        workdir: &Path,
        accessible: &Path,
        staged: Vec<StagedHookArtifact>,
        limits: ExecutionLimits,
        env_vars: HookEnvVars<'_>,
    ) -> Result<HookRuntime, RuntimeError> {
        new_with_hooks_unavailable(
            engine,
            workdir,
            accessible,
            staged,
            limits,
            env_vars,
            InferenceUnavailable::NotConfigured,
        )
        .await
    }

    /// [`new_with_hooks_full`] with the reason every `run-inference` call is answered with.
    async fn new_with_hooks_unavailable(
        engine: &wasmtime::Engine,
        workdir: &Path,
        accessible: &Path,
        staged: Vec<StagedHookArtifact>,
        limits: ExecutionLimits,
        env_vars: HookEnvVars<'_>,
        reason: InferenceUnavailable,
    ) -> Result<HookRuntime, RuntimeError> {
        HookRuntime::new(
            engine,
            &WorkdirRoot::open(workdir).unwrap(),
            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,
            Err(reason),
            None,
        )
        .await
    }

    /// Invariant 5: a pre-`on-task-start`/`on-task-end` component (all six original
    /// exports, neither task export) instantiates cleanly and is simply never dispatched
    /// for the two task events — no crash, no error logged.
    #[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"));

            // Dispatching a task event must be a silent no-op for this hook: because the
            // function is absent it is skipped before any wasm call, so no error is logged.
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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"
        );
    }

    /// A component that *does* export the two optional task functions registers them,
    /// so they can be dispatched.
    #[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"));
        });
    }

    pub(super) 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(),
            gateway: None,
        }
    }

    fn hook_log_lines(session: &Path, hook_name: &str) -> usize {
        hook_log_text(session, hook_name).lines().count()
    }

    /// The whole of one hook's error log, empty when the file was never written.
    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()
    }

    /// Simulates the `runtime.rs` dispatch sequence for a `task_acceptance: queue` launch
    /// that processes three tasks: `on-session-start` once, then a
    /// `on-task-start`/inference/`on-task-end` trio per task, then `on-session-end` once.
    ///
    /// Each hook double is bound to a single lifecycle event and its exported functions
    /// have a deliberately mismatched (no-arg) signature, so every *attempted* dispatch
    /// fails `.typed()` and appends exactly one line to that hook's error log. Counting
    /// log lines therefore counts invocations: the session-bound hooks must be hit once
    /// each while the task-bound hooks are hit once per task. Also asserts `total_turns`
    /// is the whole-launch aggregate (one per inference), which is what the once-per-launch
    /// `SessionEnd` carries.
    #[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();

        // Every double exports all eight lifecycle functions; the per-hook binding (not the
        // exports) decides which events reach it.
        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");

            // once before the loop
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    HookEvent::SessionStart,
                )
                .await;
            // three tasks, each a start / one inference / end trio
            for t in 0..3u32 {
                hooks
                    .emit(
                        &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                        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(
                        &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                        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(
                        &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                        HookEvent::TaskEnd {
                            task_id: format!("tsk_{t}"),
                            exit_status: "ok".to_string(),
                        },
                    )
                    .await;
            }
            // once after the loop, carrying the whole-launch turn aggregate
            let total_turns = hooks.total_turns();
            assert_eq!(
                total_turns, 3,
                "one turn accumulated per inference across tasks"
            );
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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"
        );
    }

    /// Invariant 6: a component missing one of the six original functions still fails
    /// `HookRuntime::new` with an error naming the missing function — the fail-fast
    /// diagnostic is not relaxed to a silent skip for the required six.
    #[test]
    fn blocking_hook_missing_required_fn_fails_new() {
        let session = TempDir::new().unwrap();
        let accessible = TempDir::new().unwrap();
        let engine = hook_test_engine();
        // Every original export except on-session-start.
        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}"
            );
        });
    }

    /// A hook component built against the *current* versioned
    /// `murmur:hook/lifecycle@0.9.0` interface (the name a freshly-compiled hook
    /// carries) instantiates and registers every required and optional function.
    /// The current versioned name is the one `resolve_lifecycle_iface` probes first.
    #[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"));
        });
    }

    /// When the versioned lifecycle instance name does not resolve (here the
    /// component exports an incompatible future version the host does not
    /// recognise), instantiation fails with the missing-export diagnostic naming
    /// the versioned interface — the probe must not swallow a genuinely absent
    /// interface.
    #[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}"
            );
        });
    }

    /// A `murmur:hook/lifecycle` double whose every export is one core function that never
    /// returns, so an epoch deadline is the only thing that can end a call to it.
    ///
    /// Unlike [`hook_double`] — whose stubs exist only to be *resolved* and would fail
    /// `TypedFunc`'s signature check if called — this one declares the real
    /// `func(session-context) -> result<hook-output, string>` type, which is what lets
    /// `call_typed` get as far as `call_async`. Only `on-session-start` is ever dispatched
    /// against it here; the other five names reuse the same function purely to satisfy
    /// `resolve_hook_fns`, exactly as `hook_double` does.
    ///
    /// The core function is lifted with 10 flat i32 params (four strings plus one
    /// `list<string>`, two i32 each) and returns an i32 pointer to the result area, since
    /// `result<hook-output, string>` exceeds one flat result. It spins before touching any
    /// of that, so the return area is never written.
    ///
    /// The exported instance must re-export every *named* type its signature references
    /// (`message`, `tool-manifest`, `hook-output`, `session-context`) — a component-model
    /// validity rule, not decoration. Omitting them fails `Component::new` with the
    /// distinctly unhelpful "instance not valid to be used as export". Structural types
    /// like `string` need no such export, which is why [`hook_double`] gets away without
    /// any of this.
    fn hook_spin_double(engine: &wasmtime::Engine) -> Component {
        hook_spin_double_iface(engine, LIFECYCLE_IFACE)
    }

    /// [`hook_spin_double`] under an explicit lifecycle instance name, so the
    /// same real-signature `on-session-start` can be presented as either
    /// lifecycle version.
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// Wrap a hand-written core module in the current lifecycle component shell.
    ///
    /// `core_body` must export `memory`, `realloc`, `oncompact` (lifted as
    /// `on-compaction`) and `noop` (every other required export). Splitting this out
    /// keeps the compaction doubles below down to the wasm that actually differs
    /// between them; the surrounding type/instance section is a component-model
    /// validity requirement, not test-specific detail (see
    /// [`hook_spin_double_iface`]).
    ///
    /// `oncompact`'s flat signature is fixed by the canonical ABI lowering of the
    /// 5-field `compaction-event`: `(list<message> → i32 i32) (u64) (f64)
    /// (option<string> → i32 i32 i32) (option<string> → i32 i32 i32)`.
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// Sentinel content an echo double reports when the option field it was asked to
    /// echo arrived as `none`. Deliberately not a plausible model name or prompt, so
    /// "the host sent none" can never be mistaken for "the host sent the string
    /// `none`".
    const MODEL_ABSENT_SENTINEL: &str = "<<absent>>";

    /// A compaction double that reports back what it saw in `compaction-event.model`:
    /// it returns `ok(replace-context([{role: "model", content: <the model string>}]))`,
    /// substituting [`MODEL_ABSENT_SENTINEL`] when the option arrived as `none`.
    fn hook_compaction_echo_model_double(engine: &wasmtime::Engine) -> Component {
        hook_compaction_echo_double(engine, "model", "model")
    }

    /// The `system-prompt` twin of [`hook_compaction_echo_model_double`] — echoes
    /// `compaction-event.system-prompt` back as the returned message's content, under
    /// role `"system-prompt"`.
    fn hook_compaction_echo_system_prompt_double(engine: &wasmtime::Engine) -> Component {
        hook_compaction_echo_double(engine, "sp", "system-prompt")
    }

    /// Shared body of the two echo doubles above: echo the `option<string>` carried by
    /// the `$<field>-some/-ptr/-len` flattened params back out as the single returned
    /// message's content, tagged with `role`.
    ///
    /// Echoing the *received* pointer/length straight back into the returned message is
    /// what makes this a real assertion on the wire value rather than on host-side Rust
    /// state — the string the test reads is the one the guest was actually handed.
    ///
    /// Return area is laid out by hand at offset 128:
    /// `result` discriminant `0` (ok); `hook-output` discriminant `1`
    /// (`replace-context`) at 132 with its `list<message>` (ptr 160, len 1) at 136/140;
    /// the one `message` at 160. `message` is 40 bytes at align 4 —
    /// `{role ptr 160/len 164, content ptr 168/len 172, id 176..188,
    /// source-id 188..200}` — each `option<string>` being a 12-byte
    /// `{discriminant, ptr, len}`. Both option discriminants are stored as `0`
    /// (`none`); the host never lifts their payload words. The role string lives
    /// at 240 and the absent-sentinel at 264, clear of the widened record.
    /// Unlike the other doubles this one needs a genuine bump `realloc` — a fixed
    /// address would lower the messages list, the model string and the system-prompt
    /// string all on top of each other, and the echoed bytes would be garbage.
    pub(super) 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)
    }

    /// A compaction double whose `on-compaction` unconditionally returns
    /// `err("boom")` — the "declared compaction hook, no safety net behind it" case.
    /// Return area at 128: `result` discriminant `1` (err), payload string
    /// (ptr 300, len 4) at 132/136.
    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)
    }

    /// Drive one `dispatch_compaction` against a single named double and hand the
    /// caller the raw result to assert on.
    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,
                    inserted_by: None,
                }],
                1234,
                0.98,
                model,
                system_prompt,
            )
            .await
    }

    /// Scenario 1: `inference.compaction.model` set in the manifest reaches the hook
    /// verbatim as `compaction-event.model`.
    #[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");
    }

    /// `inference.compaction.system_prompt` set in the manifest reaches the hook
    /// verbatim as `compaction-event.system-prompt` — and does so independently of
    /// `model`, which is left unset here.
    #[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);
    }

    /// No `system_prompt:` in the manifest arrives as `option::none` — nothing on this
    /// path substitutes a default prompt. `model` is set here to prove the two fields
    /// resolve independently.
    #[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);
    }

    /// Scenario 2: no `model:` in the manifest arrives as `option::none` — nothing on
    /// this path substitutes the primary `inference.model` first.
    #[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);
    }

    /// Scenario 4: a bound compaction hook that returns `Err` surfaces as `Err` to the
    /// caller — distinct from the `Ok(None)` that "no hook was bound" produces — while
    /// still logging to the hook's own log exactly as before.
    #[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"
        );
    }

    /// Scenario 3: no compaction-bound hook at all stays `Ok(None)` — the caller
    /// continues without compaction rather than failing the session.
    #[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,
                    inserted_by: 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"
        );
    }

    /// A hook built against the current lifecycle version receives `on-compaction`
    /// with the full 5-field record. Nothing is logged, so the dispatch really
    /// completed rather than failing `.typed()` and being isolated.
    #[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"
        );
    }

    /// The host keeps no compatibility fallback. A hook still compiled against a
    /// retired lifecycle version does not resolve at all: it fails at instantiation
    /// with the missing-export diagnostic, naming the one accepted interface and
    /// pointing the author at a rebuild, rather than being quietly accepted and
    /// then mis-dispatched.
    #[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}"
            );
        });
    }

    // ---- shell-event: `binary` (added by the `@0.5.0` bump) ----

    /// A double whose `on-shell` declares the **current** `shell-event`, and which verifies
    /// what it was handed rather than merely accepting it: the core function byte-compares
    /// the incoming `binary` string against `expect_binary` and traps on any difference. So
    /// "zero lines in the hook log" here means the guest received exactly the value the host
    /// claims it sent — a weaker double could pass while `binary` carried garbage.
    ///
    /// `shell-event` flattens to 22 core values, past the canonical ABI's 16-parameter
    /// limit, so the record arrives **indirectly**: one `i32` pointing at the lowered record
    /// in guest memory. Its field offsets are `turn` 0, `binary` ptr/len 4/8, `command`
    /// ptr/len 12/16, `argv` ptr/len 20/24, `script` disc/ptr/len 28/32/36, `recipe`
    /// disc/ptr/len 40/44/48, `outcome` disc 56. The `realloc` bump-allocates because the
    /// host lowers the record and every string it points at into that memory.
    ///
    /// The expected bytes live at offset 0 and the `result<hook-output, string>` return
    /// area at 128, so `expect_binary` must stay well under 128 bytes.
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// Stage `component` as the sole `on-shell` hook and dispatch one shell event
    /// carrying `binary`.
    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(
                &crate::workdir_fs::WorkdirRoot::open(session).unwrap(),
                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;
    }

    /// A current-version hook receives the current `shell-event`, and the `binary` it
    /// observes is byte-for-byte what the host sent — the whole point of the `@0.5.0`
    /// bump. The double traps on any mismatch, so an empty log is the assertion.
    #[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"
        );
    }

    /// Guards the test above: the double really does inspect `binary`, so a wrong value
    /// traps rather than passing silently. Without this, `current_hook_receives_the_invoked_binary_in_shell_event` would
    /// still pass if the host sent an empty or stale `binary`.
    #[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}"
        );
    }

    /// End-to-end through the real host import: a hook component that imports
    /// `murmur:runtime/inference@0.4.0` and calls `run-inference` gets back the
    /// driver double's completion text, which it returns as an `on-inference`
    /// artifact so the test can read it.
    #[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: WorkdirRoot::open(accessible.path()).unwrap(),
            inference_env: Vec::new(),
            capability_policy: crate::types::CapabilityPolicy::default(),
            network_allow_rules: Vec::new(),
            driver_grant: None,
            gateway: None,
            formation: 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,
                &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                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(),
                Ok(Arc::clone(&ctx)),
                None,
            )
            .await
            .expect("a hook importing murmur:runtime/inference must link");
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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");
    }

    /// Same component, but the capsule has no inference driver staged. The
    /// import still links (so the hook runs at all) and `run-inference` returns
    /// a clear `err` naming the manifest key to add — never a panic or an
    /// instantiation failure.
    #[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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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
        );
    }

    /// Same component, under `inference.transport: process`. The call is answered with an `err`
    /// that names the transport, not the manifest key a process capsule already declares.
    #[test]
    fn hook_run_inference_under_process_transport_names_the_transport() {
        let session = TempDir::new().unwrap();
        let accessible = TempDir::new().unwrap();
        let engine = hook_test_engine();

        let rt = tokio::runtime::Runtime::new().unwrap();
        let (artifacts, records, refusals) = rt.block_on(async {
            let mut hooks = new_with_hooks_unavailable(
                &engine,
                session.path(),
                accessible.path(),
                vec![staged_double_named(
                    "caller",
                    HookBinding::OnInference,
                    hook_inference_caller_double(&engine),
                )],
                ExecutionLimits::default(),
                HookEnvVars::default(),
                InferenceUnavailable::ProcessTransport,
            )
            .await
            .expect("the import is defined under process transport, so the hook links");
            let artifacts = hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    inference_event(),
                )
                .await;
            (
                artifacts,
                hooks.drain_inference_records(),
                hooks.drain_spend_refusals(),
            )
        });

        assert_eq!(artifacts.len(), 1);
        let payload = &artifacts[0].payload;
        assert_eq!(payload, InferenceUnavailable::ProcessTransport.message());
        assert!(
            payload.contains("inference.transport: process"),
            "got: {payload}"
        );
        assert!(
            !payload.contains("add inference.driver.artifact"),
            "got: {payload}"
        );
        assert!(records.is_empty(), "{records:?}");
        assert!(refusals.is_empty(), "{refusals:?}");
    }

    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,
        }
    }

    /// A current-version lifecycle double that *imports* `murmur:runtime/inference@0.4.0`
    /// and, on `on-inference`, calls `run-inference` with an empty message list
    /// and `model: none`, then returns whichever string the call produced —
    /// the completion text on success, the error message on failure — as
    /// `hook-output::artifact`. `on-inference` is the one event whose artifact
    /// `dispatch` forwards, which is what makes the result observable.
    ///
    /// Memory and `realloc` live in a separate core module so the lowered import
    /// can reference them without a cyclic instantiation. `result<inference-response,
    /// string>` lays `ok`/`err` payloads at the same offset (record align 8 →
    /// discriminant at 0, payload at 8), so one code path copies either string's
    /// `(ptr, len)` into the artifact.
    pub(super) 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 (enum "replace-context" "seed-context"))
    (export "context-insertion" (type (eq 1)))
    (type (option 2))
    (type (record
      (field "role" string)
      (field "content" string)
      (field "id" 0)
      (field "source-id" 0)
      (field "inserted-by" 3)))
    (export "message" (type (eq 4)))
    (type (list 5))
    (type (record
      (field "messages" 6)
      (field "system-prompt" 0)
      (field "model" 0)))
    (export "inference-request" (type (eq 7)))
    (type (record
      (field "text" string)
      (field "model-used" string)
      (field "input-tokens" u64)
      (field "output-tokens" u64)))
    (export "inference-response" (type (eq 9)))
    (type (result 10 (error string)))
    (export "run-inference" (func (param "request" 8) (result 11)))
  ))
  (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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// A hook that never returns from `on-session-start` is cut off at its epoch deadline
    /// and flows through the *existing* per-hook isolation: the failure is logged to
    /// `logs/hook-<name>.log` by `log_hook_error` and `dispatch` carries on to the next
    /// hook rather than aborting the session. A spinning hook is bounded, not unkillable.
    #[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();
        // Without a running ticker the epoch never advances and the deadline cannot fire.
        let _ticker = crate::limits::EpochTicker::spawn(&engine);

        let limits = ExecutionLimits {
            deadline_seconds: 1,
            ..ExecutionLimits::default()
        };

        // The spinner is registered first, so a second hook receiving the same event at all
        // proves dispatch resumed after the trap instead of unwinding.
        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");

            // Returns at all == the spinning guest was interrupted; without the deadline the
            // await below would never complete.
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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}"
        );

        // The bystander's own call fails TypedFunc's signature check (its stub is a bare
        // `func() -> ()`), which is itself an ordinary hook error — the point is only that
        // dispatch reached it at all after the spinner trapped.
        assert!(
            session
                .path()
                .join("logs")
                .join("hook-bystander.log")
                .exists(),
            "dispatch must continue to the next hook after a deadline trap"
        );
    }

    // ── murmur:runtime/tokens: the ungated host counter ──────────────────────

    use crate::tokens_import::TOKENS_IFACE_VERSIONED;

    /// The string [`hook_token_counter_double`] measures. Long enough that the
    /// `text.len() / 4` fallback in [`crate::agent::count_tokens`] and a real
    /// `cl100k_base` count cannot coincide by accident.
    const TOKEN_PROBE: &str =
        "the quick brown fox jumps over the lazy dog, and then counts its own tokens";

    /// A lifecycle double that *imports* `murmur:runtime/tokens@0.4.0` and, on
    /// `on-session-start`, calls `count` on [`TOKEN_PROBE`] and returns the result rendered
    /// as decimal digits in `err(...)`. A returned `err` is appended verbatim to
    /// `logs/hook-<name>.log`, so the number the host computed is readable from the log.
    ///
    /// Memory and `realloc` live in a separate core module so the lowered import can
    /// reference them without a cyclic instantiation, the same arrangement
    /// [`hook_inference_caller_double`] uses. `count` lowers to `(param i32 i32) (result
    /// i64)` — the string's `(ptr,len)`, and the `u64` return as one flat value.
    ///
    /// Return area at 128: `result` discriminant `1` (err); the string `(ptr,len)` at
    /// 132/136, its bytes written from 400 upwards.
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// Invariant: `murmur:runtime/tokens@0.4.0#count` is registered on the main hook
    /// linker with no capability grant, and it answers with the host's own count.
    ///
    /// The double is staged default-deny — no network rules, no filesystem scope, no
    /// task-io grant — so it links only because the counter is ungated. The number it
    /// reports is compared against [`crate::agent::count_tokens`] directly: a host that
    /// registered some other counter would link fine and still fail here.
    #[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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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")
        );
    }

    /// The staging linker registers the counter too. `on-stage` runs on a linker of its
    /// own, built before any inference driver or task exists; a hook that imports `count`
    /// and is dispatched there must link rather than fail instantiation.
    #[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,
            &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
            &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
            &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
            std::slice::from_ref(&staged),
            Vec::new(),
            &HookEnvVars::default(),
            ExecutionLimits::default(),
        )
        .expect("on-stage dispatch itself never fails on a hook-returned error");

        // The double exports no `on-stage`, so dispatch fails looking that export up.
        // Reaching *that* error means instantiation and lifecycle resolution both
        // succeeded, which is the fact under test — a linker missing the counter would
        // instead log a failure to instantiate.
        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"
        );
    }

    // ── murmur:task-io/read: the per-hook grant and the scope table ───────────

    use crate::task_io_import::test_support::{probe_double, reader_double, REPORT_SEP};

    /// A staged hook with the task-io grant flipped on, everything else default-deny — no
    /// network rules and no filesystem scope, so anything it reports about the task can only
    /// have come through the import.
    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,
                filesystem_records: false,
                records_hooks_dir: None,
                task_io_read,
                conversation_read: false,
                state_store: None,
                state_dir: None,
                config_json: None,
            },
            ..staged_double_named(name, binding, component)
        }
    }

    /// Every line the probe double logged, in dispatch order.
    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()
    }

    /// The scope table in `wit/hook/deps/murmur-task-io/read.wit`, driven through the real
    /// dispatch paths in the order `runtime.rs` uses them: staging, session start, task start,
    /// the agent loop, task end, session end.
    ///
    /// `on-stage` runs on its own linker and its own throwaway state; `on-task-start` fires
    /// before the task enters scope and `on-session-end` after it leaves, so all three report
    /// `no-task` (`!1`) even though the hook is granted. Mid-loop the input is readable and the
    /// output is `no-output` (`!2`); at `on-task-end` both are readable.
    #[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);

        // `on-stage` is dispatched on its own linker with its own fresh state.
        let mut on_stage = staged.clone();
        on_stage.config.binding = HookBinding::OnStage;
        dispatch_stage(
            &engine,
            &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
            &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
            &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
            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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    HookEvent::SessionStart,
                )
                .await;
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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;

            // What `run_task_with_reopens` does around one attempt of the agent loop.
            hooks.begin_task_attempt("task text".to_string(), "task text".to_string());
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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"
        );
    }

    /// A hook without `capabilities.task_io.read` still links and is still dispatched; every
    /// read returns `not-granted` (`!0`) rather than trapping, and no task text reaches it.
    #[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"
        );
    }

    /// The `reopen-task` a granted `on-task-end` hook returns carries exactly what it read —
    /// both input forms and the output — proving the arm and the import compose.
    #[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());
        });
    }

    // ── murmur:conversation/read: the per-hook grant and the paged record ─────

    use crate::conversation_import::test_support::{
        reader_double as conversation_reader_double, REPORT_SEP as CONVERSATION_SEP,
    };

    /// A staged hook with the conversation grant flipped on or off, everything else default-deny:
    /// no network rules and no filesystem scope, so anything it reports about the record can only
    /// have come through the import.
    fn staged_conversation_hook(
        component: Component,
        conversation_read: bool,
    ) -> StagedHookArtifact {
        StagedHookArtifact {
            grant: HookCapabilityGrant {
                conversation_read,
                ..HookCapabilityGrant::default()
            },
            ..staged_double_named("reader", HookBinding::OnTaskEnd, component)
        }
    }

    /// The id the nth message of a hand-written record carries.
    fn recorded_id(n: usize) -> String {
        format!("msg_{n:032x}")
    }

    /// Write `count` messages into the record `root`/`context_id` holds, ids `msg_…01` upward.
    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();
    }

    /// Dispatch the reader double once at `on-task-end` and return the two pages it reported.
    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,
                &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                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(),
                Err(InferenceUnavailable::NotConfigured),
                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()
        })
    }

    /// A granted hook reads a record an earlier run left: newest first, two per page, with the
    /// cursor of the first page continuing into the second without repeating or skipping one.
    #[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"
        );
    }

    /// The last page of a record says so: `next-cursor` is `none` once the oldest message has
    /// been returned, which is what ends a paging loop.
    #[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]);
    }

    /// An ungranted hook still links, still runs and still ends its task; the read is a named
    /// error it can branch on, and it is distinguishable from the empty page a session with no
    /// record returns.
    #[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"]);
    }

    /// A session the runtime built with no record — `context.record: off`, or the `process`
    /// transport, which never builds a message list — reads as an empty page rather than an
    /// error.
    #[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]
        );
    }

    /// A record holding a malformed line returns every other message and reports the bad one,
    /// naming the 1-based line number: one bad line never fails a read.
    #[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,
                &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                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(),
                Err(InferenceUnavailable::NotConfigured),
                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}"
        );
    }

    // ── Per-hook capability grants (default-deny network + filesystem) ────────

    use crate::network_policy::RequestTarget;
    use murmur_artifact::{Capabilities, FilesystemCapabilities, NetworkCapabilities};

    /// Build the grant a hook entry declaring exactly these two sub-blocks would get,
    /// going through the same `HookCapabilityGrant::derive` the staging path uses.
    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(),
                records: false,
            }),
            shell: None,
            spawn: None,
            env: None,
            limits: None,
            resources: None,
            state: None,
            task_io: None,
            conversation: None,
            plan: None,
            install: None,
            containment: None,
        };
        HookCapabilityGrant::derive(Some(&caps), "test-capsule").expect("grant is valid")
    }

    /// A hook store built exactly as the three instantiation sites build one, so the
    /// network suite exercises the real `WasiHttpView` wiring rather than a stand-in.
    fn hook_store_state(root: &Path, grant: &HookCapabilityGrant) -> HookStoreState {
        hook_store_state_with_gateway(root, grant, None)
    }

    /// [`hook_store_state`] for a hook whose entry declares `gateway:`.
    fn hook_store_state_with_gateway(
        root: &Path,
        grant: &HookCapabilityGrant,
        gateway: Option<Arc<crate::credential_gateway::CredentialGateway>>,
    ) -> HookStoreState {
        HookStoreState {
            limits: ExecutionLimits::default().limiter(),
            table: ResourceTable::new(),
            wasi: build_wasi_ctx(root, &HookEnvVars::default(), grant, gateway.as_deref())
                .expect("wasi ctx builds"),
            http: WasiHttpCtx::new(),
            http_hooks: NetworkPolicyHooks {
                network_allow_rules: grant.network_allow_rules.clone(),
                gateway,
                formation: None,
                task_provenance: None,
            },
        }
    }

    /// Whether a hook store's own HTTP gate admits a request for `uri`. Nothing is sent.
    fn hook_store_admits(state: &HookStoreState, uri: &str) -> bool {
        state
            .http_hooks
            .admit(&uri.parse().expect("uri parses"))
            .is_ok()
    }

    /// A hook whose entry declares `gateway:` has its gateway-addressed request sent to the
    /// gateway's upstream on an empty allow-list, carrying the rendered key in place of a forged
    /// one, while a request to any host its grant does not name is still denied.
    #[test]
    fn hook_store_with_a_gateway_reaches_its_upstream_and_still_denies_an_unlisted_host() {
        use std::io::Read;

        use http_body_util::{BodyExt, Empty};

        use crate::credential_gateway::{CredentialGateway, GatewayMetering};

        const KEY: &str = "sk-hook-gateway-marker";
        let upstream = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        upstream.set_nonblocking(true).unwrap();
        let gateway = Arc::new(
            CredentialGateway::new(
                "grafana",
                &format!("http://{}/otlp", upstream.local_addr().unwrap()),
                murmur_artifact::UpstreamAuth {
                    header: "Authorization".to_string(),
                    value: "Bearer {key}".to_string(),
                },
                Some(Arc::new(
                    crate::gateway_credential::GatewayCredential::resolve(
                        "grafana",
                        &murmur_artifact::ApiKeyReference::Literal(KEY.to_string()),
                        None,
                    )
                    .unwrap(),
                )),
                GatewayMetering::Unmetered,
            )
            .unwrap(),
        );
        let root = TempDir::new().unwrap();
        let grant = HookCapabilityGrant::default();
        let mut state = hook_store_state_with_gateway(root.path(), &grant, Some(gateway));

        // As wasi-http hands it over: `host` set to the gateway authority the guest addressed.
        let request = http::Request::builder()
            .method("POST")
            .uri("http://127.0.0.1:9/otlp/v1/traces")
            .header("host", "127.0.0.1:9")
            .header("authorization", "Bearer forged")
            .body(
                Empty::<bytes::Bytes>::new()
                    .map_err(|err| match err {})
                    .boxed_unsync(),
            )
            .unwrap();
        let rt = tokio::runtime::Runtime::new().unwrap();
        let head = rt.block_on(async {
            // wasi-http drives the response future on a task of its own.
            let sent = tokio::spawn(Box::into_pin(state.http().hooks.send_request(
                request,
                None,
                Box::new(async { Ok(()) }),
            )));
            let mut head = String::new();
            for _ in 0..100 {
                if let Ok((mut stream, _)) = upstream.accept() {
                    stream.set_nonblocking(false).unwrap();
                    stream
                        .set_read_timeout(Some(Duration::from_millis(500)))
                        .unwrap();
                    let mut buffer = [0u8; 4096];
                    let read = stream.read(&mut buffer).unwrap_or(0);
                    head = String::from_utf8_lossy(&buffer[..read]).into_owned();
                    break;
                }
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
            sent.abort();
            head
        });
        let head = head.to_ascii_lowercase();
        assert!(head.starts_with("post /otlp/v1/traces "), "{head}");
        assert_eq!(head.matches("authorization:").count(), 1, "{head}");
        assert!(
            head.contains(&format!("authorization: bearer {KEY}")),
            "{head}"
        );
        assert_eq!(head.matches("\r\nhost:").count(), 1, "{head}");
        assert!(
            head.contains(&format!("\r\nhost: {}\r\n", upstream.local_addr().unwrap())),
            "the upstream, not the gateway authority, is named: {head}"
        );

        assert!(!hook_store_admits(&state, "http://127.0.0.1:1/unlisted"));
        assert!(!hook_store_admits(
            &state,
            "https://telemetry.example.com/ingest"
        ));
    }

    /// Default-deny, network half: a hook whose operator entry declared no `capabilities:`
    /// has an empty allow-list, and its store's own HTTP gate denies every request.
    #[test]
    fn ungranted_hook_store_denies_every_outbound_request() {
        let root = TempDir::new().unwrap();
        let grant = HookCapabilityGrant::default();
        let state = hook_store_state(root.path(), &grant);

        assert!(!hook_store_admits(
            &state,
            "https://telemetry.example.com/ingest"
        ));
        assert!(!hook_store_admits(&state, "http://127.0.0.1:1/local"));
    }

    /// Granted network: exactly the declared host is admitted; every other host is denied
    /// by the same `NetworkPolicyHooks` capsules and tools use.
    #[test]
    fn granted_hook_store_admits_only_the_declared_host() {
        let root = TempDir::new().unwrap();
        // A loopback port nothing listens on: the policy decision is observable without
        // the test ever completing a connection.
        let grant = grant_of(Some("http://127.0.0.1:1"), None);
        let state = hook_store_state(root.path(), &grant);

        assert!(
            hook_store_admits(&state, "http://127.0.0.1:1/ingest"),
            "the granted host must pass the allow-list"
        );
        assert!(
            !hook_store_admits(&state, "http://127.0.0.1:2/ingest"),
            "a different port on the granted host must still be denied"
        );
        assert!(
            !hook_store_admits(&state, "https://evil.example.com/x"),
            "an undeclared host must be denied"
        );
    }

    /// Default-deny, filesystem half: nothing is preopened. Observable because
    /// `preopened_dir` requires the directory to exist — building a context rooted at a
    /// path that does not exist can only succeed if no preopen was attempted at all.
    #[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(),
            None,
        )
        .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"
        );
        // The default grant names no store, so there is no second preopen to add.
        assert!(HookCapabilityGrant::default().state_dir.is_none());
    }

    /// Both filesystem axes at once: the granted project subtree *and* the durable store. Two
    /// preopens, mounted at two guest paths, from two host trees that need not be related.
    #[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, None)
            .expect("both preopens are built");

        // The scoped subtree is created under the hook's own root.
        assert!(root.path().join("hook-state").is_dir());
        // The store is a separate host tree, untouched and outside that root.
        assert!(!state.path().starts_with(root.path()));
        assert_eq!(
            std::fs::read_to_string(state.path().join("cursor.json")).unwrap(),
            "{\"seen\":7}"
        );
    }

    /// A hook may hold durable state without being handed the project directory: exactly one
    /// preopen, and it is the store. Observable because `preopened_dir` requires its target to
    /// exist — a context rooted at a path that does not exist can only build if the workdir
    /// preopen was never attempted.
    #[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, None)
            .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"
        );
    }

    /// The state preopen is really attempted, not merely configured: `preopened_dir` requires its
    /// target to exist, so a store directory that is missing fails the build. This is what makes
    /// the two tests above assertions about preopens rather than about `Ok(())`.
    ///
    /// The staging path creates the store before building any context, so this state is
    /// unreachable in a real run — it is here to pin the observation the other tests rest on.
    #[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, None) {
            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}"
        );
    }

    /// Granted filesystem: exactly one directory — `<root>/<scope>` — is preopened, and it
    /// is created if absent. Sibling paths under the same root are never preopened, so a
    /// guest has no descriptor with which to name them.
    #[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, None)
            .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");
        // Writes inside the scope land on the real host filesystem at that path.
        std::fs::write(scoped.join("cursor.json"), b"{}").unwrap();
        assert!(scoped.join("cursor.json").exists());
        // Nothing else under the root was touched or mounted.
        assert!(sibling.is_dir());
        assert!(root.path().join("secret.txt").exists());
    }

    /// An existing scope directory is reused as-is: granting a scope must not clobber
    /// state a previous run left there.
    #[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")),
            None,
        )
        .unwrap();

        assert_eq!(
            std::fs::read_to_string(scoped.join("cursor.json")).unwrap(),
            "{\"seen\":7}"
        );
    }

    /// A scope that cannot be created fails instantiation rather than silently degrading
    /// to "no filesystem access" — an operator who declared a scope must not have it
    /// quietly dropped.
    #[test]
    fn unusable_scope_fails_instantiation() {
        let root = TempDir::new().unwrap();
        // A regular file where the scope directory would go: create_dir_all cannot succeed.
        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")),
            None,
        ) {
            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}"
        );
    }

    /// No exempt instantiation path — blocking hooks. `HookRuntime::new` applies the grant
    /// to the *project* directory, and only the granted subtree there.
    #[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"
        );
    }

    /// No exempt instantiation path — blocking hooks, default-deny. Instantiation succeeds
    /// against a project directory that does not exist, which is only possible because no
    /// preopen is attempted.
    #[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());
    }

    /// No exempt instantiation path — `on-stage`. `call_stage_once` builds its context from
    /// the same grant; the dispatch itself fails on the stub's signature, which is after
    /// the WASI context has already been built.
    #[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,
                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
                &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
                &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"
        );
    }

    /// No exempt instantiation path — async hooks. The single instance built for an async
    /// hook at `HookRuntime::new` applies the grant carried on its `StagedHookArtifact`,
    /// under the project dir, exactly as the blocking path does: how often the grant is
    /// applied differs between the two paths, whether it is applied does not.
    #[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"
        );
    }

    /// No self-escalation: the grant is derived from the *operator's* manifest entry, so a
    /// hook artifact whose own bundled murmur.yaml declares broad capabilities is still
    /// fully denied when the operator granted it nothing.
    #[test]
    fn hook_self_declared_capabilities_do_not_grant_anything() {
        // The hook artifact's own bundled manifest, self-declaring broad access.
        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);

        // The operator's own manifest, granting this hook nothing.
        let operator_manifest = murmur_artifact::RuntimeManifest::from_yaml_str(
            r#"
name: cap
version: 0.0.1
artifacts:
  - name: local/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 state = hook_store_state(root.path(), &grant);
        assert!(!hook_store_admits(&state, "https://evil.example.com/exfil"));
        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)));
    }

    // ── Uniform hook-output dispatch: unsupported-arm faults ──────────────────

    /// An `on-tool-call` double returning `ok(<arm>)`, where `arm_disc` selects the
    /// `hook-output` variant: `0` = `none`, `1` = `replace-context([])`, `2` =
    /// `write-manifests([])`. `on-tool-call` honors no arm (see
    /// [`HONORED_OUTPUT_ARM`]), so `1` and `2` are unsupported-arm faults while `0`
    /// is silent. The list arms return an empty list — `(ptr,len)` left at `(0,0)` —
    /// so no guest heap layout is needed.
    ///
    /// Return area at 128: `result` discriminant `0` (ok); `hook-output`
    /// discriminant at 132; list `(ptr,len)` at 136/140.
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// A *current-version* (`@0.9.0`, 7-case `hook-output`) `on-task-end` double that
    /// returns `ok(<arm>)`. `arm_disc` selects the variant: `0` = `none`, `4` =
    /// `reopen-task(reason)`. The `reopen-task` payload is a static string at guest
    /// offset 300 so the host lifts the real bytes the guest declared, exercising the
    /// typed `reopen-task` wire path end-to-end rather than host-side Rust state.
    ///
    /// Return area at 128: `result` discriminant `0` (ok); `hook-output` discriminant
    /// at 132; the `reopen-task` string `(ptr,len)` at 136/140. `on-task-end`'s flat
    /// params are the two strings of `task-end-event` (4 i32).
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// Scenario: a blocking `on-task-end` hook returning `reopen-task(reason)` is the one
    /// honored arm for that event; `dispatch_task_end` surfaces it as a
    /// [`TaskReopen`] naming the hook and carrying the exact feedback string the guest
    /// returned, with no dispatch fault logged.
    #[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();
        // arm 4 = reopen-task.
        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"
        );
    }

    /// An `on-task-end` hook returning `none` (arm 0) drives no reopen:
    /// `dispatch_task_end` returns `None`.
    #[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());
        });
    }

    /// Invariant: `reopen-task` returned from any event other than `on-task-end` is a
    /// dispatch fault, not a silent grant. Bind the same 7-case double to `on-tool-call`
    /// (its `on-tool-call` export is a bare stub, so dispatching it fails `.typed()` —
    /// but the point tested here is the honored-arm *table*: `on-tool-call` honors no
    /// arm, so even if it returned `reopen-task` it would be classified `Fault`, exactly
    /// as `classify_output` decides generically).
    #[test]
    fn reopen_task_is_only_honored_at_on_task_end() {
        // Pure table check: reopen-task is honored ONLY by 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"
            );
        }
    }

    // ---- task-start-event: the new u64 fields, and the seed-context arm ----

    /// An `on-task-start` double that checks what it was handed and reports back with a
    /// selectable `hook-output` arm.
    ///
    /// It traps unless the `context-window` it received equals `expect_window`, so a test
    /// asserting "no log line" is asserting the guest saw the host's real number rather
    /// than merely that the call completed. `arm_disc` selects the returned `hook-output`
    /// case: `0` = `none`, `5` = `seed-context([one message])`.
    ///
    /// `on-task-start`'s flat params are the canonical lowering of the 7-field
    /// `task-start-event`: three `string`s (6 i32) then `input-bytes`, `budget-tokens`,
    /// `context-window`, `prior-tokens` (4 i64).
    ///
    /// Return area at 128: `result` discriminant `0` (ok); `hook-output` discriminant at
    /// 132; the `list<message>` `(ptr,len)` = `(160,1)` at 136/140. The one `message`
    /// occupies 160..200 — `role` ptr/len at 160/164, `content` ptr/len at 168/172, and
    /// the two `option<string>` discriminants at 176 and 188, both `0` (`none`).
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// Dispatch one `on-task-start` to `double` under the name `starter`, with
    /// `context_window` on the event, and hand back the buffered dispatch faults.
    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(
                &crate::workdir_fs::WorkdirRoot::open(session).unwrap(),
                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()
    }

    /// [`dispatch_task_start_against`] through the honoring path: hands back both the
    /// proposed seed and the buffered 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())
    }

    /// Invariant: the `context-window` a hook sees at `on-task-start` is the number the
    /// host put on the event. The double traps on any other value, so an empty log is
    /// the assertion.
    #[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"
        );
    }

    /// A capsule with no `context:` block sends `0`, and the guest sees `0` — not a
    /// stale value and not an absent field.
    #[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);
    }

    /// Invariant: `on-task-start` honors `seed-context`. The hook's messages come back on
    /// the [`HookSeed`], no fault is raised, and nothing reaches the hook's error log —
    /// the failure this test exists to catch is the seed being discarded silently.
    #[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);
    }

    /// A hook returning `none` from `on-task-start` proposes nothing. This is what makes a
    /// bound-but-silent hook inert.
    #[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());
    }

    /// `on-task-start` is the one event that honors `seed-context`. Reads the honored-arm
    /// table directly, so widening the arm to a second event fails here rather than in
    /// whichever dispatch test happens to cover it.
    #[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"
            );
        }
    }

    /// An `on-stage` double returning `ok(<arm>)`, `arm_disc` as in
    /// [`hook_tool_call_arm_double`]. `on-stage` honors only `write-manifests`
    /// (`2`), so `1` (`replace-context`) is an unsupported-arm fault and `0`
    /// (`none`) is silent — exercised through `dispatch_stage`.
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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(),
        }
    }

    /// Scenario (a): a blocking hook that returns a non-`none` arm the event does not
    /// honor (`on-tool-call` → `write-manifests`) gets exactly one line in
    /// `logs/hook-<name>.log` naming the hook/event/arm, and exactly one entry from
    /// the new drain accessor carrying the same structured fields. The session
    /// continues — `emit` returns normally with no artifact.
    #[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();
        // arm 2 = write-manifests, unsupported for on-tool-call.
        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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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");

            // Draining is destructive: a second drain yields nothing.
            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}");
    }

    /// Scenario (c): `none` from an event that honors no arm produces no fault —
    /// no log line, no buffered fault.
    #[test]
    fn none_from_blocking_hook_produces_no_fault() {
        let session = TempDir::new().unwrap();
        let accessible = TempDir::new().unwrap();
        let engine = hook_test_engine();
        // arm 0 = none.
        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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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"
        );
    }

    /// Scenario (b), on-inference/artifact half: the honored arm is committed and
    /// produces no fault. Uses the inference-caller double (returns `artifact`).
    #[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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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"
        );
    }

    /// Scenario (b), on-compaction/replace-context half: the honored arm still
    /// produces `Ok(Some(messages))` and buffers no fault.
    #[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,
                        inserted_by: 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"
        );
    }

    /// Scenario (d): an `on-stage` hook returning an unsupported arm
    /// (`replace-context`) is logged via `dispatch_stage`'s path, and no
    /// `trace.jsonl` is written — `dispatch_stage` runs before the trace writer
    /// exists.
    #[test]
    fn on_stage_unsupported_arm_is_logged_not_traced() {
        let workdir = TempDir::new().unwrap();
        let engine = hook_test_engine();
        // arm 1 = replace-context, unsupported for on-stage.
        let staged = staged_double_named(
            "stager",
            HookBinding::OnStage,
            hook_stage_arm_double(&engine, 1),
        );

        dispatch_stage(
            &engine,
            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
            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"
        );
    }

    /// Scenario (d) control: `write-manifests` from `on-stage` is the honored arm —
    /// it writes the manifest and logs no fault.
    #[test]
    fn on_stage_write_manifests_remains_honored() {
        let workdir = TempDir::new().unwrap();
        let engine = hook_test_engine();
        // arm 2 = write-manifests (honored) — but with an empty list, so nothing is
        // written; the point is that no fault is logged for the honored arm.
        let staged = staged_double_named(
            "stager",
            HookBinding::OnStage,
            hook_stage_arm_double(&engine, 2),
        );

        dispatch_stage(
            &engine,
            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
            &crate::workdir_fs::WorkdirRoot::open(workdir.path()).unwrap(),
            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"
        );
    }

    /// Scenario (e): an async hook returning a non-`none` arm the event does not honor is
    /// logged exactly once to `logs/hook-<name>.log` *and* recorded as a dispatch fault, so
    /// it reaches `trace.jsonl` as `hook_dispatch_error` on the same terms a blocking hook's
    /// fault does. The drain awaits the worker directly, so nothing is left in flight to
    /// race with the assertion.
    #[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();
        // arm 2 = write-manifests, unsupported for on-tool-call.
        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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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");
    }

    /// The honored-arm table has exactly one entry per WIT lifecycle function, and
    /// every entry's arm (when set) is a real `hook-output` variant name. Guards the
    /// single-source-of-truth invariant against a typo or a missing/extra row.
    #[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"
                );
            }
        }
    }

    /// Cross-crate invariant: the runtime's honored-arm table and the manifest parser's
    /// `commit_policy_for_binding` (which `parse_hook_config_from_yaml` validates a hook's
    /// declared `commit_policy` against at staging time) must describe the same thing. If
    /// one table gains, loses, or renames an entry without the other, this fails here
    /// rather than as a silent mid-session `hook_dispatch_error`.
    #[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 {
                // The one pair that cannot agree, by construction: `on-inference` honors
                // the `artifact` arm, but `HookCommitPolicy` has no `artifact` spelling —
                // so no non-`none` policy is declarable for this binding, and the parser
                // rejects every one.
                assert_eq!(arm, Some("artifact"), "{event}");
                assert_eq!(decision_arm, None, "{event}");
                assert_eq!(policy, None, "{event}");
                continue;
            }

            // An event honors an arm at exactly one of its two phases, so the union is the
            // whole of what a hook may declare for that binding. The two gated events sit on
            // the `Decide` side (`deny`); every other event sits on the `Observe` side.
            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);

            // The arm the runtime commits is exactly the policy a hook may declare, and an
            // event that honors nothing admits no policy but `none`.
            assert_eq!(
                honored,
                policy.as_ref().map(HookCommitPolicy::as_str),
                "{event}: honored arm and declarable commit_policy disagree"
            );
        }

        // Spelled out rather than left to the loop: a table edit that silently dropped one of
        // these two rows would still pass the union check above.
        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);
    }

    /// `deny` is honored at exactly one place, and the tables are what say so: from the
    /// observation dispatch of any of the nine events it is an unsupported arm, including the
    /// two events whose decision point does honor it.
    #[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"
            );
        }
    }

    /// The decision point honors `deny` and nothing else: `none` is silent, and every other
    /// arm is a fault there too.
    #[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")
            ));
        }
    }

    /// A policy hook can only subtract. [`CallDecision`] has exactly two variants, `Proceed`
    /// carries nothing and `Denied` carries only the hook's name and its reason — no
    /// capability, path, grant or policy value — so a decision that widens the manifest's
    /// grant is not expressible. The exhaustive destructuring match is the assertion: adding a
    /// third variant, or a payload field to `Denied`, stops this compiling.
    #[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");
                }
            }
        }
    }

    /// `gates_calls` is the branch that decides whether the agent loop resolves a call at all.
    /// It is `true` only for a blocking hook declaring `commit_policy: deny` on one of the two
    /// gated bindings: an observer bound to the same event does not turn it on, and neither
    /// does a `deny` the parser would have refused on an ungated binding.
    #[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}");
        }
    }

    // ── Async hooks: one reused instance, one serialized queue, a guaranteed drain ──

    /// Run `body` on a current-thread runtime inside a `LocalSet` — the shape the session's
    /// agent loop runs in. Async hook workers are `spawn_local` tasks, so `HookRuntime::new`
    /// needs this context whenever a staged hook is async.
    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)
    }

    /// A staged `execution_mode: async` hook with the default (deny-everything) grant and the
    /// default `on_overflow: drop`.
    fn staged_async(name: &str, binding: HookBinding, component: Component) -> StagedHookArtifact {
        staged_async_granted(
            name,
            binding,
            component,
            HookCapabilityGrant::default(),
            HookOverflowPolicy::Drop,
        )
    }

    /// [`staged_async`] with an explicit grant and overflow policy.
    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,
            gateway: None,
        }
    }

    /// Guest address of [`hook_marker_double`]'s history buffer. Above the fixed address its
    /// `realloc` hands out, so lowering an event's strings into the guest cannot overwrite the
    /// history the test reads back.
    const MARKER_BUFFER: i32 = 4096;

    /// A double that records its own call history in guest memory and reports it once, at the
    /// end.
    ///
    /// Each call appends one byte to a buffer held in the instance's linear memory — `S` for
    /// `on-session-start`, `T` for `on-tool-call`, `E` for `on-session-end` — and returns
    /// `ok(none)`. `on-session-end` additionally returns `err(<the whole buffer>)`, which
    /// `log_hook_error` writes verbatim as the single line of `logs/hook-<name>.log`.
    ///
    /// So one log line answers three questions at once: whether the same instance (and
    /// therefore the same linear memory) was reused across events, in what order the calls
    /// actually ran, and how many of them ran at all. A hook re-instantiated per event would
    /// report a one-character history; a reordered or interleaved queue would spell a
    /// different string.
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// The single line a [`hook_marker_double`] wrote: its whole call history.
    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()
    }

    /// Invariant: an async hook is instantiated once and reused, so state it accumulates in
    /// its linear memory survives across lifecycle events. Two *different* events reach the
    /// same instance; the history it reports at session end spells both, which a per-event
    /// re-instantiation could not produce.
    #[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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    HookEvent::SessionStart,
                )
                .await;
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    tool_call_event(),
                )
                .await;
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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"
        );
    }

    /// Invariant: one queue, one worker, one call at a time — the order the hook observes is
    /// the order the agent loop dispatched in. The dispatch calls below return without ever
    /// yielding to the worker, so every job is queued before any of them runs.
    #[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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    tool_call_event(),
                )
                .await;
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    HookEvent::SessionStart,
                )
                .await;
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    tool_call_event(),
                )
                .await;
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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"
        );
    }

    /// A double that reports the first environment variable of its WASI context back to the
    /// host as its `on-session-start` error string, so the test reads the value the *guest*
    /// actually sees rather than a host-side stand-in.
    ///
    /// [`build_wasi_ctx`] injects the grant's `MURMUR_ARTIFACT_CONFIG` first and
    /// `MURMUR_OTEL_ENDPOINT` after it, so entry 0 is the config JSON whenever the hook's entry
    /// declared one, and the otel endpoint otherwise. An empty environment reports `no-env`,
    /// which is why a hook handed `HookEnvVars::default()` and an ungranted
    /// `HookCapabilityGrant` cannot be mistaken for one handed the real thing.
    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 $context-insertion (enum "replace-context" "seed-context"))
  (type $message (record
    (field "role" string)
    (field "content" string)
    (field "id" (option string))
    (field "source-id" (option string))
    (field "inserted-by" (option $context-insertion))))
  (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 "context-insertion" (type $context-insertion))
    (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")
    }

    /// Invariant: an async hook is built with the session's real [`HookEnvVars`], not
    /// `HookEnvVars::default()`. This is what makes an async OTel exporter possible at all —
    /// without it the hook never learns its endpoint.
    #[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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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"
        );
    }

    /// The first environment entry [`hook_env_probe_double`] observes, for a hook staged with
    /// `grant` and a session carrying `env_vars`. `no-env` when the environment is empty.
    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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    HookEvent::SessionStart,
                )
                .await;
            hooks.drain_async_hooks().await;
        });

        marker_history(session.path(), "probe")
    }

    /// A hook's `config:` block reaches its guest as `MURMUR_ARTIFACT_CONFIG`, read off the
    /// hook's own grant rather than off the session-wide [`HookEnvVars`] — which is what scopes
    /// it to the declaring hook.
    #[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"}"#
        );
    }

    /// Default-deny, environment half: a hook whose entry declared neither config nor a session
    /// variable observes an empty environment.
    #[test]
    fn an_unconfigured_hook_sees_no_environment_at_all() {
        assert_eq!(
            first_hook_env_entry(HookCapabilityGrant::default(), HookEnvVars::default()),
            "no-env"
        );
    }

    /// Config is injected first, so the runtime-owned `MURMUR_*` variables are applied after it
    /// and keep precedence: whatever an operator writes in a `config:` block, it cannot shadow a
    /// value the runtime owns.
    #[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"
        );
    }

    /// A member session's hooks see its formation id as `MURMUR_FORMATION_ID`, from the session.
    #[test]
    fn a_member_sessions_hook_sees_its_formation_id() {
        let id = crate::formation::FormationId::mint();
        assert_eq!(
            first_hook_env_entry(
                HookCapabilityGrant::default(),
                HookEnvVars {
                    formation_id: Some(id.as_str()),
                    ..HookEnvVars::default()
                },
            ),
            id.as_str()
        );
    }

    /// The formation id comes from the session and never from the process environment, so a
    /// session in no formation hands its hooks none even when the process carries the variable.
    #[test]
    fn a_hook_never_sees_a_formation_id_from_the_process_environment() {
        let _guard = crate::formation::FORMATION_ENV_LOCK
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let id = crate::formation::FormationId::mint();
        std::env::set_var(crate::formation::FORMATION_ID_ENV, id.as_str());
        let seen = first_hook_env_entry(HookCapabilityGrant::default(), HookEnvVars::default());
        std::env::remove_var(crate::formation::FORMATION_ID_ENV);
        assert_eq!(seen, "no-env");
    }

    /// A member's hooks see the members it may call as `MURMUR_FORMATION_PEERS`, at their
    /// virtual addresses, from the session, and never from the process environment.
    #[test]
    fn a_members_hook_sees_its_formation_peers_and_only_from_the_session() {
        let _guard = crate::formation::FORMATION_ENV_LOCK
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let authority = crate::formation_credentials::FormationAuthority::for_test();
        let member = Arc::new(crate::formation_credentials::FormationMember::from_bundle(
            authority.member_bundle("planner", &["coder", "reviewer"]),
        ));
        let seen = first_hook_env_entry(
            HookCapabilityGrant::default(),
            HookEnvVars {
                formation_member: Some(&member),
                ..HookEnvVars::default()
            },
        );
        assert_eq!(
            seen,
            "coder=http://coder.formation.invalid reviewer=http://reviewer.formation.invalid"
        );

        std::env::set_var(
            crate::formation::FORMATION_PEERS_ENV,
            "decoy=http://localhost:1",
        );
        let outside = first_hook_env_entry(HookCapabilityGrant::default(), HookEnvVars::default());
        std::env::remove_var(crate::formation::FORMATION_PEERS_ENV);
        assert_eq!(outside, "no-env");
    }

    /// Invariant: work queued by the last `emit` of a session finishes *before* the
    /// `LocalSet` that owns the workers is torn down.
    ///
    /// Deliberately shaped like `runtime.rs`: everything happens inside
    /// `local.run_until(...)`, and the assertion runs after that future has returned and
    /// `local` has been dropped, matching how a real session tears down.
    #[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(
                            &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                            HookEvent::SessionStart,
                        )
                        .await;
                    hooks
                        .emit(
                            &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                            tool_call_event(),
                        )
                        .await;
                    hooks
                        .emit(
                            &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                            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"
        );
    }

    /// Invariant: the drain is bounded and reports what it abandoned. The spinner's first
    /// queued call burns past the budget (only its epoch deadline can end it), so the second
    /// is still outstanding when the budget expires: the worker is abandoned and the hook is
    /// named in the log and in a `hook_dispatch_error`-bound fault rather than silently lost.
    #[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();
        // Without a running ticker the epoch never advances and the spinner never returns
        // at all — the drain's own timer cannot preempt guest code on this thread.
        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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    HookEvent::SessionStart,
                )
                .await;
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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}"
        );
    }

    /// Invariant: `on_overflow: drop` (the default) never blocks the agent loop. The
    /// dispatches below never yield, so the worker cannot drain anything while they run: the
    /// queue fills at [`ASYNC_HOOK_QUEUE_DEPTH`] and every event past that is discarded,
    /// counted, and reported per event.
    #[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(
                        &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                        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");
    }

    /// Invariant: `on_overflow: block` loses nothing. Past the queue's depth the dispatch
    /// call waits for the worker to make room instead of discarding, so the hook observes
    /// every event — which the history it reports at session end counts exactly.
    #[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(
                        &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                        tool_call_event(),
                    )
                    .await;
            }
            hooks
                .emit(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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]);
    }

    /// Invariant: an async hook's genuine `Err` reaches `trace.jsonl` as a
    /// `hook_dispatch_error`, not just `logs/hook-<name>.log`. The marker double's
    /// `on-session-end` returns `err(<history>)`, which is the async equivalent of any hook
    /// call that fails outright.
    #[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(
                    &crate::workdir_fs::WorkdirRoot::open(session.path()).unwrap(),
                    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"
        );
    }

    /// Invariant: an async hook never commits output. `replace-context` is the arm
    /// `on-compaction` *does* honor for a blocking hook;
    /// from an async hook it is discarded silently — no replacement, and not a fault either,
    /// since the hook did nothing wrong.
    #[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,
                        inserted_by: 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:?}");
    }
}