capsule-runtime 0.11.0

//! Per-guest execution limits: epoch-based deadlines and store resource caps.
//!
//! Every guest this runtime executes — capsule `run`, tool/driver `run`, and each hook
//! lifecycle call — runs against a `Store` wired with two independent bounds:
//!
//! * an **epoch deadline**, set before each invocation and enforced by [`EpochTicker`],
//!   which advances the engine's epoch on a fixed interval so a guest that never returns
//!   traps instead of hanging the session; and
//! * a **resource limiter** ([`ExecutionLimiter`]) capping linear-memory growth, table
//!   growth, and instance count.
//!
//! Both are sourced from the manifest's `capabilities.limits` block, falling back to the
//! `DEFAULT_*` constants below for any field it omits. A silent manifest means defaults,
//! never "unlimited".
//!
//! ## Two mechanisms for one deadline
//!
//! Epoch checks are compiled into wasm loop back-edges and function entries, so the epoch
//! deadline can only fire while guest code is executing. A call blocked inside a host call — a
//! `wasi:clocks` sleep, a `wasi:io/poll` on an HTTP response that never comes — is out of its
//! reach until control returns to wasm.
//!
//! Every component call is therefore also wrapped in a wall-clock bound, [`bound_call`], of
//! `deadline_seconds` plus [`WALL_DEADLINE_GRACE`], measured per phase (instantiation, then the
//! call) from the moment that phase's future starts. A guest computing past its deadline is
//! trapped by the epoch first, with the epoch's message; the wall bound fires only where the
//! epoch cannot, and stops the call by dropping its future, which disposes the guest fiber and
//! any outbound request pending under it. Each firing is pushed into the session's
//! [`CallDeadlines`] latch, which the trace drains into `call_deadline` lines.

use std::{
    future::Future,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc, Mutex,
    },
    thread,
    time::{Duration, Instant},
};

use wasmtime::{Engine, ResourceLimiter, StoreLimits, StoreLimitsBuilder};

/// How often [`EpochTicker`] advances the engine epoch. Also the granularity of every
/// deadline: a budget is rounded down to a whole number of ticks (minimum one).
pub(crate) const EPOCH_TICK_INTERVAL: Duration = Duration::from_millis(100);

/// Default cap on guest linear-memory growth — 512 MiB. Chosen to sit far above what any
/// real capsule, tool, or hook needs while still bounding a runaway allocation loop.
pub const DEFAULT_MEMORY_BYTES: usize = 512 * 1024 * 1024;

/// Default cap on guest table growth, in elements.
pub const DEFAULT_TABLE_ELEMENTS: usize = 100_000;

/// Default cap on instances a single store may create. A component instantiates one core
/// instance per core module, so single digits is typical; this only stops a pathological
/// instantiation loop.
pub const DEFAULT_INSTANCES: usize = 1_000;

/// What the wall-clock bound on a component call adds to `deadline_seconds`.
///
/// Long enough that a guest computing past its deadline is still caught by the epoch trap,
/// which needs a tick of [`EPOCH_TICK_INTERVAL`] and control back in wasm, so only a call
/// pending in the host is stopped by the wall bound.
pub(crate) const WALL_DEADLINE_GRACE: Duration = Duration::from_secs(2);

const _: () = assert!(
    WALL_DEADLINE_GRACE.as_millis() > EPOCH_TICK_INTERVAL.as_millis(),
    "the wall bound must leave the epoch at least one tick to fire first"
);

/// Default wall-clock budget for a single guest invocation — 10 minutes.
///
/// Deliberately generous: the budget covers host-call time as well as guest compute (see
/// the module docs), so a capsule whose `run` drives a long agent loop through host
/// `invoke` calls must not trip it. Anything wanting a tight bound sets
/// `capabilities.limits.deadline_seconds` explicitly.
pub const DEFAULT_DEADLINE_SECONDS: u64 = 600;

/// Default wall-clock budget for a single *hook lifecycle call* — 30 seconds.
///
/// An order of magnitude below [`DEFAULT_DEADLINE_SECONDS`], and deliberately so: a hook call
/// is one bounded piece of work (an HTTP POST to a collector, a scorer invocation), not a
/// whole agent loop, so the capsule-wide default would let a wedged hook stall the session for
/// most of ten minutes *per event*. Applies to blocking and async hooks alike, and only when
/// the operator left `capabilities.limits.deadline_seconds` unset — an explicit value is
/// honored verbatim for every guest, hooks included. See [`ExecutionLimits::for_hook_calls`].
pub const HOOK_DEFAULT_DEADLINE_SECONDS: u64 = 30;

const _: () = assert!(
    HOOK_DEFAULT_DEADLINE_SECONDS < DEFAULT_DEADLINE_SECONDS,
    "a hook call must not be able to stall a session for the capsule-wide budget"
);

/// Fully-resolved execution limits for one session's guests: the manifest's
/// `capabilities.limits` block with every omitted field replaced by its `DEFAULT_*`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ExecutionLimits {
    /// Cap on linear-memory growth, in bytes.
    pub memory_bytes: usize,
    /// Cap on table growth, in elements.
    pub table_elements: usize,
    /// Cap on instances a single store may create.
    pub instances: usize,
    /// Wall-clock budget for a single guest invocation, in seconds.
    pub deadline_seconds: u64,
}

impl Default for ExecutionLimits {
    fn default() -> Self {
        Self {
            memory_bytes: DEFAULT_MEMORY_BYTES,
            table_elements: DEFAULT_TABLE_ELEMENTS,
            instances: DEFAULT_INSTANCES,
            deadline_seconds: DEFAULT_DEADLINE_SECONDS,
        }
    }
}

impl ExecutionLimits {
    /// Resolve a manifest `capabilities.limits` block, substituting the default for each
    /// field the manifest left out (and for the whole block when it is absent).
    #[must_use]
    pub fn resolve(declared: Option<&murmur_artifact::ResourceLimits>) -> Self {
        let defaults = Self::default();
        let Some(declared) = declared else {
            return defaults;
        };
        Self {
            memory_bytes: declared.memory_bytes.unwrap_or(defaults.memory_bytes),
            table_elements: declared.table_elements.unwrap_or(defaults.table_elements),
            instances: declared.instances.unwrap_or(defaults.instances),
            deadline_seconds: declared
                .deadline_seconds
                .unwrap_or(defaults.deadline_seconds),
        }
    }

    /// This limit block re-flavored for hook lifecycle calls: every resource cap unchanged,
    /// but the wall-clock budget falls back to [`HOOK_DEFAULT_DEADLINE_SECONDS`] instead of
    /// [`DEFAULT_DEADLINE_SECONDS`] when the operator declared none.
    ///
    /// Takes the *undefaulted* `capabilities.limits.deadline_seconds` rather than reading it
    /// back off `self`, because that is the only way to tell "the operator set nothing" from
    /// "the operator explicitly set 600" — the second must be honored verbatim, hooks
    /// included. Callers get it from [`crate::CapabilityPolicy::hook_limits`].
    #[must_use]
    pub fn for_hook_calls(self, declared_deadline_seconds: Option<u64>) -> Self {
        Self {
            deadline_seconds: declared_deadline_seconds.unwrap_or(HOOK_DEFAULT_DEADLINE_SECONDS),
            ..self
        }
    }

    /// These hook limits for one hook whose own `artifacts:` entry declares
    /// `capabilities.limits.deadline_seconds` as `entry_deadline_seconds`, which wins over the
    /// session's hook deadline. Every resource cap stays capsule-wide.
    #[must_use]
    pub(crate) fn for_hook_entry(self, entry_deadline_seconds: Option<u64>) -> Self {
        Self {
            deadline_seconds: entry_deadline_seconds.unwrap_or(self.deadline_seconds),
            ..self
        }
    }

    /// This budget expressed in epoch ticks, for `Store::set_epoch_deadline`.
    ///
    /// Rounds down to whole ticks but never to zero — a deadline of zero ticks would trap
    /// the guest before it executed a single instruction.
    pub(crate) fn deadline_ticks(&self) -> u64 {
        let tick_ms = EPOCH_TICK_INTERVAL.as_millis() as u64;
        (self.deadline_seconds.saturating_mul(1_000) / tick_ms).max(1)
    }

    /// The wall-clock bound on one phase of a call under these limits: `deadline_seconds` plus
    /// [`WALL_DEADLINE_GRACE`], handed to [`bound_call`] or [`CallDeadlines::bound`].
    pub(crate) fn wall_bound(&self) -> Duration {
        Duration::from_secs(self.deadline_seconds).saturating_add(WALL_DEADLINE_GRACE)
    }

    /// Build the store limiter enforcing these limits.
    pub(crate) fn limiter(&self) -> ExecutionLimiter {
        ExecutionLimiter::new(*self)
    }
}

/// `ResourceLimiter` enforcing an [`ExecutionLimits`] on one store.
///
/// Wraps `wasmtime::StoreLimits` purely to reuse its comparison logic, built with
/// `trap_on_grow_failure(false)` so a denial reaches us as `Ok(false)` instead of
/// wasmtime's own untyped `bail!`. We then convert the denial into an error carrying a
/// message that names the manifest field responsible, and record it.
///
/// The recorded flag — not the error type — is what [`classify_guest_failure`] keys on.
/// wasmtime raises `Trap::AllocationTooLarge` only from its GC and libcall paths, never
/// from a limiter denial, so downcasting the resulting error to `Trap` cannot identify
/// this case and matching on message text would be brittle.
pub(crate) struct ExecutionLimiter {
    inner: StoreLimits,
    limits: ExecutionLimits,
    /// Set the first time this limiter denies a growth request.
    denial: Option<String>,
}

impl ExecutionLimiter {
    fn new(limits: ExecutionLimits) -> Self {
        let inner = StoreLimitsBuilder::new()
            .memory_size(limits.memory_bytes)
            .table_elements(limits.table_elements)
            .instances(limits.instances)
            .trap_on_grow_failure(false)
            .build();
        Self {
            inner,
            limits,
            denial: None,
        }
    }

    /// The limits this limiter enforces.
    pub(crate) fn limits(&self) -> ExecutionLimits {
        self.limits
    }

    /// The message from the first growth request this limiter denied, if any.
    pub(crate) fn denial(&self) -> Option<&str> {
        self.denial.as_deref()
    }
}

impl ResourceLimiter for ExecutionLimiter {
    fn memory_growing(
        &mut self,
        current: usize,
        desired: usize,
        maximum: Option<usize>,
    ) -> wasmtime::Result<bool> {
        if self.inner.memory_growing(current, desired, maximum)? {
            return Ok(true);
        }
        let message = format!(
            "linear memory growth to {desired} bytes exceeds \
             capabilities.limits.memory_bytes ({})",
            self.limits.memory_bytes
        );
        self.denial = Some(message.clone());
        Err(wasmtime::Error::msg(message))
    }

    fn table_growing(
        &mut self,
        current: usize,
        desired: usize,
        maximum: Option<usize>,
    ) -> wasmtime::Result<bool> {
        if self.inner.table_growing(current, desired, maximum)? {
            return Ok(true);
        }
        let message = format!(
            "table growth to {desired} elements exceeds \
             capabilities.limits.table_elements ({})",
            self.limits.table_elements
        );
        self.denial = Some(message.clone());
        Err(wasmtime::Error::msg(message))
    }

    fn instances(&self) -> usize {
        self.inner.instances()
    }

    fn tables(&self) -> usize {
        self.inner.tables()
    }

    fn memories(&self) -> usize {
        self.inner.memories()
    }
}

/// Advances `Engine::increment_epoch` every [`EPOCH_TICK_INTERVAL`] for the lifetime of a
/// session's engine. Without a running ticker an epoch deadline can never fire, so exactly
/// one of these is spawned per `Engine` in `stage_session` and its guard is held on
/// `StagedSession` — a `wasmtime::Engine` is an `Arc` handle, so that single ticker reaches
/// every capsule, tool, and hook store cloned from it.
pub(crate) struct EpochTicker {
    stop: Arc<AtomicBool>,
}

impl EpochTicker {
    /// Spawn the ticker thread for `engine`.
    ///
    /// A plain OS thread rather than a tokio task: staging dispatches `on-stage` hooks on
    /// its own current-thread runtime while the rest of the session uses a multi-thread
    /// one, and the ticker must keep running across both regardless of which (if any)
    /// runtime is entered.
    ///
    /// The thread holds only an `EngineWeak`, so it can never keep the engine alive and
    /// exits on its own once the last `Engine` handle drops. The stop flag covers the
    /// reverse case — the guard dropping while cloned `Engine` handles still live in store
    /// states — so the two together bound the thread's life from both ends.
    pub(crate) fn spawn(engine: &Engine) -> Self {
        let weak = engine.weak();
        let stop = Arc::new(AtomicBool::new(false));
        let thread_stop = Arc::clone(&stop);

        thread::spawn(move || {
            while !thread_stop.load(Ordering::Relaxed) {
                thread::sleep(EPOCH_TICK_INTERVAL);
                match weak.upgrade() {
                    Some(engine) => engine.increment_epoch(),
                    None => break,
                }
            }
        });

        Self { stop }
    }
}

impl Drop for EpochTicker {
    /// Signals the ticker thread to exit. Deliberately does not join: the thread notices
    /// the flag within one tick and exits on its own, and blocking session teardown for up
    /// to `EPOCH_TICK_INTERVAL` to watch it happen buys nothing.
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
    }
}

/// A component call the wall-clock bound stopped, with how long it had run when it was stopped.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct WallDeadline {
    pub(crate) elapsed: Duration,
}

/// Await `call` for at most `bound`, measured from now. Past it the future is dropped, which
/// is what stops a component call pending in a host wait, and the result is a [`WallDeadline`].
///
/// The one wall-clock bound every component entry point goes through, by way of
/// [`CallDeadlines::bound`]. Needs a tokio runtime with its time driver enabled.
pub(crate) async fn bound_call<F: Future>(
    bound: Duration,
    call: F,
) -> Result<F::Output, WallDeadline> {
    let started = Instant::now();
    tokio::time::timeout(bound, call)
        .await
        .map_err(|_| WallDeadline {
            elapsed: started.elapsed(),
        })
}

/// What a component is running as when its call is bound, as `call_deadline.role` names it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub(crate) enum ComponentRole {
    /// A script capsule's own component.
    Capsule,
    /// An inference driver run for a model call: the agent loop's turn or a hook's
    /// `run-inference`.
    Driver,
    /// Every other tool-world component call, a driver reached by name included.
    Tool,
    Hook,
    ProcessDriver,
}

impl ComponentRole {
    /// The value `call_deadline.role` carries.
    pub(crate) fn wire_name(self) -> &'static str {
        match self {
            Self::Capsule => "capsule",
            Self::Driver => "driver",
            Self::Tool => "tool",
            Self::Hook => "hook",
            Self::ProcessDriver => "process_driver",
        }
    }
}

/// One firing of the wall-clock bound, as its `call_deadline` line records it.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub(crate) struct CallDeadline {
    /// The tool name of the artifact, the hook's name, or the capsule's manifest `name`.
    pub(crate) component: String,
    pub(crate) role: ComponentRole,
    /// `instantiate`, `run`, a hook lifecycle function such as `on-session-start`, or a process
    /// driver function: `describe`, `launch`, `parse` or `classify-exit`.
    pub(crate) call: String,
    pub(crate) deadline_seconds: u64,
    pub(crate) elapsed_ms: u64,
}

/// The session's record of every component call the wall-clock bound stopped, filled where no
/// trace writer is in reach and drained by the one that is.
///
/// Created at staging and cloned into everything that runs a component: the capsule store and
/// every tool or driver store dispatched from it, each hook store, the process driver. The
/// trace writer holding a clone drains it ahead of every line it writes, so each firing becomes
/// exactly one `call_deadline` line before the record of the call it cut and before
/// `session_end`.
#[derive(Debug, Clone, Default)]
pub(crate) struct CallDeadlines(Arc<Mutex<Vec<CallDeadline>>>);

impl CallDeadlines {
    /// [`bound_call`] for one phase of a call under `limits`, recording a firing.
    ///
    /// A bound that fires inside another — a hook's `run-inference` driver call, a tool a script
    /// capsule's `run` invokes — records itself; the outer one never fires for it, since the
    /// inner call has returned. An outer bound that fires first drops the inner call before it
    /// can record.
    pub(crate) async fn bound<F: Future>(
        &self,
        component: &str,
        role: ComponentRole,
        call: &str,
        limits: &ExecutionLimits,
        future: F,
    ) -> Result<F::Output, WallDeadline> {
        let deadline = (limits.deadline_seconds, limits.wall_bound());
        self.bound_within(component, role, call, deadline, future)
            .await
    }

    /// [`Self::bound`] with the wall bound given outright beside the `deadline_seconds` it is
    /// recorded under, so a test need not wait out a whole second.
    async fn bound_within<F: Future>(
        &self,
        component: &str,
        role: ComponentRole,
        call: &str,
        (deadline_seconds, bound): (u64, Duration),
        future: F,
    ) -> Result<F::Output, WallDeadline> {
        let bounded = bound_call(bound, future).await;
        if let Err(cut) = &bounded {
            self.lock().push(CallDeadline {
                component: component.to_string(),
                role,
                call: call.to_string(),
                deadline_seconds,
                elapsed_ms: u64::try_from(cut.elapsed.as_millis()).unwrap_or(u64::MAX),
            });
        }
        bounded
    }

    /// Records `cut`, a firing a guest worker's own bound recorded for a call it ran.
    pub(crate) fn record(&self, cut: CallDeadline) {
        self.lock().push(cut);
    }

    /// Takes every firing recorded since the last take, oldest first.
    pub(crate) fn take(&self) -> Vec<CallDeadline> {
        std::mem::take(&mut *self.lock())
    }

    /// Records `cut` as if a bound had fired, for a test of what drains the latch.
    #[cfg(test)]
    pub(crate) fn push_for_test(&self, cut: CallDeadline) {
        self.lock().push(cut);
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, Vec<CallDeadline>> {
        self.0
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }
}

/// The failure text of a call `subject` the wall-clock bound stopped, on the paths that report
/// failures as plain text. The epoch's wording is [`GuestFailure::message`].
pub(crate) fn wall_deadline_message(subject: &str, seconds: u64) -> String {
    format!(
        "{subject} waited on the host past its {seconds}s execution deadline \
         (capabilities.limits.deadline_seconds) and was stopped"
    )
}

/// [`wall_deadline_message`] for hook `hook`'s `fn_name` call, naming the key on the hook's own
/// `artifacts:` entry, `entry`, as [`GuestFailure::hook_message`] does.
pub(crate) fn wall_deadline_hook_message(
    hook: &str,
    entry: &str,
    fn_name: &str,
    seconds: u64,
) -> String {
    format!(
        "hook '{hook}' {fn_name} waited on the host past its {seconds}s execution deadline \
         (capabilities.limits.deadline_seconds on the artifacts: entry '{entry}') and was stopped"
    )
}

/// Why a guest invocation failed, separating the two limits this module enforces from an
/// ordinary trap.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum GuestFailure {
    /// The epoch deadline fired — the guest ran longer than `deadline_seconds`.
    DeadlineExceeded { seconds: u64 },
    /// The store's resource limiter denied a growth request.
    ResourceLimit { message: String },
    /// Anything else: guest panic, `unreachable`, host error.
    Other,
}

/// The single classification point for every failed guest invocation — capsule `run`,
/// tool/driver `run`, and hook lifecycle calls all route their `Err` through here so the
/// three outcomes stay distinguishable without each call site re-deriving the logic.
///
/// A limiter denial is checked first and wins over any trap code, because the denial is
/// recorded at the moment of refusal and is therefore authoritative about the cause; the
/// error wasmtime surfaces afterwards may have been reshaped on its way out of the guest.
pub(crate) fn classify_guest_failure(
    error: &wasmtime::Error,
    limiter: &ExecutionLimiter,
) -> GuestFailure {
    if let Some(message) = limiter.denial() {
        return GuestFailure::ResourceLimit {
            message: message.to_string(),
        };
    }
    if matches!(
        error.downcast_ref::<wasmtime::Trap>(),
        Some(wasmtime::Trap::Interrupt)
    ) {
        return GuestFailure::DeadlineExceeded {
            seconds: limiter.limits().deadline_seconds,
        };
    }
    GuestFailure::Other
}

impl GuestFailure {
    /// Render this failure for the `String`-returning tool/driver and hook paths, which
    /// report failures as plain text rather than a [`crate::errors::RuntimeError`].
    ///
    /// `subject` names the guest, e.g. `tool 'echo'` or `hook 'audit'`. The [`Self::Other`]
    /// wording is the pre-existing generic phrasing, kept verbatim so only the two limit
    /// cases read differently.
    pub(crate) fn message(&self, subject: &str, error: &wasmtime::Error) -> String {
        match self {
            Self::DeadlineExceeded { seconds } => format!(
                "{subject} exceeded its {seconds}s execution deadline \
                 (capabilities.limits.deadline_seconds) and was interrupted"
            ),
            Self::ResourceLimit { message } => {
                format!("{subject} exceeded its configured resource limits: {message}")
            }
            Self::Other => format!("{subject} trapped: {error}"),
        }
    }

    /// [`Self::message`] for hook `hook`'s `fn_name` call. A deadline names the key on the hook's
    /// own `artifacts:` entry, `entry`, which sets that hook's deadline ahead of the capsule-wide
    /// one.
    pub(crate) fn hook_message(
        &self,
        hook: &str,
        entry: &str,
        fn_name: &str,
        error: &wasmtime::Error,
    ) -> String {
        let subject = format!("hook '{hook}' {fn_name}");
        match self {
            Self::DeadlineExceeded { seconds } => format!(
                "{subject} exceeded its {seconds}s execution deadline \
                 (capabilities.limits.deadline_seconds on the artifacts: entry '{entry}') and was \
                 interrupted"
            ),
            _ => self.message(&subject, error),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn resolve_uses_defaults_when_manifest_is_silent() {
        assert_eq!(ExecutionLimits::resolve(None), ExecutionLimits::default());
    }

    #[test]
    fn resolve_fills_each_omitted_field_independently() {
        let declared = murmur_artifact::ResourceLimits {
            memory_bytes: Some(1024),
            table_elements: None,
            instances: None,
            deadline_seconds: Some(5),
        };
        let limits = ExecutionLimits::resolve(Some(&declared));

        assert_eq!(limits.memory_bytes, 1024);
        assert_eq!(limits.deadline_seconds, 5);
        assert_eq!(limits.table_elements, DEFAULT_TABLE_ELEMENTS);
        assert_eq!(limits.instances, DEFAULT_INSTANCES);
    }

    /// A hook call gets the lower default when the manifest declared no deadline, while every
    /// resource cap stays exactly what the capsule-wide resolution produced.
    #[test]
    fn hook_calls_get_the_lower_deadline_default_when_none_is_declared() {
        let capsule = ExecutionLimits::resolve(None);
        let hook = capsule.for_hook_calls(None);

        assert_eq!(capsule.deadline_seconds, DEFAULT_DEADLINE_SECONDS);
        assert_eq!(hook.deadline_seconds, HOOK_DEFAULT_DEADLINE_SECONDS);
        assert_eq!(hook.memory_bytes, capsule.memory_bytes);
        assert_eq!(hook.table_elements, capsule.table_elements);
        assert_eq!(hook.instances, capsule.instances);
    }

    /// An explicitly declared deadline wins for hooks too — including one that happens to
    /// equal the capsule-wide default, which is why resolution keys on the undefaulted
    /// `Option` rather than on the resolved value.
    #[test]
    fn an_explicit_deadline_is_honored_for_hook_calls() {
        let declared = murmur_artifact::ResourceLimits {
            memory_bytes: None,
            table_elements: None,
            instances: None,
            deadline_seconds: Some(DEFAULT_DEADLINE_SECONDS),
        };
        let capsule = ExecutionLimits::resolve(Some(&declared));

        assert_eq!(
            capsule
                .for_hook_calls(declared.deadline_seconds)
                .deadline_seconds,
            DEFAULT_DEADLINE_SECONDS
        );
        assert_eq!(capsule.for_hook_calls(Some(5)).deadline_seconds, 5);
    }

    #[test]
    fn deadline_ticks_converts_seconds_and_never_yields_zero() {
        let limits = ExecutionLimits {
            deadline_seconds: 1,
            ..ExecutionLimits::default()
        };
        // 1s / 100ms per tick.
        assert_eq!(limits.deadline_ticks(), 10);
        assert_eq!(ExecutionLimits::default().deadline_ticks(), 6_000);
    }

    /// A limiter that has denied a growth request classifies as `ResourceLimit` even though
    /// the resulting error is a plain message with no `Trap` to downcast to — the case the
    /// recorded-denial design exists to cover.
    #[test]
    fn classify_reports_resource_limit_after_a_denial() {
        let limits = ExecutionLimits {
            memory_bytes: 1024,
            ..ExecutionLimits::default()
        };
        let mut limiter = limits.limiter();
        let error = limiter
            .memory_growing(0, 4096, None)
            .expect_err("growth past memory_bytes must be denied");

        match classify_guest_failure(&error, &limiter) {
            GuestFailure::ResourceLimit { message } => {
                assert!(message.contains("capabilities.limits.memory_bytes"));
            }
            other => panic!("expected ResourceLimit, got {other:?}"),
        }
    }

    #[test]
    fn classify_reports_deadline_for_an_interrupt_trap() {
        let limits = ExecutionLimits {
            deadline_seconds: 7,
            ..ExecutionLimits::default()
        };
        let limiter = limits.limiter();
        let error = wasmtime::Error::from(wasmtime::Trap::Interrupt);

        assert_eq!(
            classify_guest_failure(&error, &limiter),
            GuestFailure::DeadlineExceeded { seconds: 7 }
        );
    }

    #[test]
    fn a_hook_deadline_names_the_key_on_the_hooks_entry() {
        let error = wasmtime::Error::from(wasmtime::Trap::Interrupt);
        let failure = GuestFailure::DeadlineExceeded { seconds: 2 };
        assert_eq!(
            failure.hook_message("compact", "local/compact", "on-compaction", &error),
            "hook 'compact' on-compaction exceeded its 2s execution deadline \
             (capabilities.limits.deadline_seconds on the artifacts: entry 'local/compact') and \
             was interrupted"
        );
        assert_eq!(
            failure.message("tool 'echo'", &error),
            "tool 'echo' exceeded its 2s execution deadline (capabilities.limits.deadline_seconds) \
             and was interrupted",
            "every other guest keeps the capsule-wide key"
        );
        assert_eq!(
            GuestFailure::Other.hook_message("h", "local/h", "on-stage", &error),
            GuestFailure::Other.message("hook 'h' on-stage", &error)
        );
    }

    /// An ordinary guest panic must stay in the generic bucket, and must keep the exact
    /// pre-slice wording on the string-returning paths.
    #[test]
    fn classify_reports_other_for_an_unrelated_trap() {
        let limiter = ExecutionLimits::default().limiter();
        let error = wasmtime::Error::from(wasmtime::Trap::UnreachableCodeReached);

        let failure = classify_guest_failure(&error, &limiter);
        assert_eq!(failure, GuestFailure::Other);
        assert!(failure
            .message("tool 'echo'", &error)
            .starts_with("tool 'echo' trapped: "));
    }

    #[test]
    fn the_wall_bound_is_the_deadline_plus_a_positive_grace() {
        assert!(WALL_DEADLINE_GRACE > Duration::ZERO);
        let limits = ExecutionLimits {
            deadline_seconds: 3,
            ..ExecutionLimits::default()
        };
        assert_eq!(
            limits.wall_bound(),
            Duration::from_secs(3) + WALL_DEADLINE_GRACE
        );
    }

    #[tokio::test]
    async fn bound_call_returns_a_finished_futures_output() {
        assert_eq!(
            bound_call(Duration::from_millis(50), async { 7 }).await,
            Ok(7)
        );
    }

    #[tokio::test]
    async fn bound_call_stops_a_pending_future_after_the_bound() {
        let bound = Duration::from_millis(30);
        let cut = bound_call(bound, std::future::pending::<()>())
            .await
            .expect_err("a future that never finishes must be stopped");
        assert!(cut.elapsed >= bound, "stopped after {:?}", cut.elapsed);
    }

    /// A firing is recorded once with what the call was, and a call that finished in time
    /// records nothing.
    #[tokio::test]
    async fn the_latch_records_one_entry_per_firing() {
        let deadlines = CallDeadlines::default();
        let bound = (3, Duration::from_millis(30));
        let finished = deadlines
            .bound_within("echo", ComponentRole::Tool, "run", bound, async { 1 })
            .await;
        assert_eq!(finished, Ok(1));
        assert!(deadlines.take().is_empty());

        let cut = deadlines
            .bound_within(
                "stall",
                ComponentRole::Driver,
                "run",
                bound,
                std::future::pending::<()>(),
            )
            .await;
        assert!(cut.is_err());
        let taken = deadlines.take();
        assert_eq!(taken.len(), 1);
        assert_eq!(taken[0].component, "stall");
        assert_eq!(taken[0].role.wire_name(), "driver");
        assert_eq!(taken[0].call, "run");
        assert_eq!(taken[0].deadline_seconds, 3);
        assert!(taken[0].elapsed_ms >= 30);
        assert!(deadlines.take().is_empty(), "a take clears the latch");
    }

    #[test]
    fn wall_deadline_messages_name_the_key_that_set_the_deadline() {
        assert_eq!(
            wall_deadline_message("tool 'stall'", 3),
            "tool 'stall' waited on the host past its 3s execution deadline \
             (capabilities.limits.deadline_seconds) and was stopped"
        );
        assert_eq!(
            wall_deadline_hook_message("sleep", "local/sleep", "on-session-start", 2),
            "hook 'sleep' on-session-start waited on the host past its 2s execution deadline \
             (capabilities.limits.deadline_seconds on the artifacts: entry 'local/sleep') and was \
             stopped"
        );
    }

    #[test]
    fn limiter_allows_growth_within_the_cap() {
        let limits = ExecutionLimits {
            memory_bytes: 4096,
            ..ExecutionLimits::default()
        };
        let mut limiter = limits.limiter();

        assert!(limiter.memory_growing(0, 4096, None).unwrap());
        assert!(limiter.denial().is_none());
    }
}