autumn_web/plugin_sandbox/capability/

mod.rs

//! What a granted capability actually *does*, and everything it is checked
//! against on the way (issue #1632).
//!
//! [`manifest`](super::manifest) decides what a plugin may ask for.
//! [`grants`](super::grants) decides what each grant is scoped to. This module
//! is where a request from the guest meets both, plus the quota and the audit
//! ledger, before it reaches a subsystem:
//!
//! ```text
//!  guest writes {"op":"call","call":"kv-get","key":"cart"}
//!        │
//!        ▼
//!  granted?      ── no ─► denied: capability-not-granted   (recorded)
//!        │ yes
//!        ▼
//!  in scope?     ── no ─► denied: not-in-grant             (recorded)
//!        │ yes
//!        ▼
//!  under quota?  ── no ─► denied: quota-exceeded           (recorded)
//!        │ yes
//!        ▼
//!  derive the physical name from the manifest, never from the guest
//!        │
//!        ▼
//!  backend ─────────────► {"op":"call_result","id":1,"status":"ok",…} (recorded)
//! ```
//!
//! # The guest never names a physical thing
//!
//! Every step above is ordinary defence. The step that makes cross-tenant and
//! host-table access *impossible* rather than merely refused is the second to
//! last one: the guest names a **logical** key, table, host or job type, and the
//! host derives the physical name from the manifest and the ambient tenant. A
//! guest cannot ask for another tenant's row because there is no field in the
//! wire protocol where the tenant would go. It cannot ask for a host-application
//! table because the physical name is `plugin_<plugin>_<table>` and both halves
//! come from a validated manifest. It cannot smuggle SQL because no call carries
//! any.
//!
//! That is why the adversarial corpus in
//! `autumn/tests/integration/plugin_sandbox_capabilities.rs` is mostly about
//! *spelling*: an escape here would have to be a name that survives derivation,
//! not a missing check.
//!
//! # Denials are answers, not failures
//!
//! A refused call comes back as a `call_result` the guest can read. It does not
//! trap and it does not fail the request. A plugin that hits a ceiling should
//! degrade — render the panel without the live number — and an author debugging
//! one needs to see *which* rule refused. Every denial is also recorded in the
//! outcome's ledger and logged once, which is what the operator audit surface
//! is built from.
// autumn-panic-gate: request-path module — production code path must be panic-free.
// See CONTRIBUTING.md "Request-path panic gate". Justify exceptions with
// #[allow(clippy::<lint>, reason = "…")] at the narrowest scope.
#![cfg_attr(
    not(test),
    deny(
        clippy::unwrap_used,
        clippy::expect_used,
        clippy::panic,
        clippy::unreachable,
        clippy::todo,
        clippy::unimplemented,
        clippy::indexing_slicing,
        clippy::string_slice,
        clippy::arithmetic_side_effects,
    )
)]

pub mod audit;
pub mod db;
pub mod jobs;
pub mod kv;
pub mod outbound;
pub mod quota;
pub mod render;

use std::collections::BTreeMap;
use std::fmt;
use std::sync::Arc;

use serde::{Deserialize, Serialize};

use super::grants::CapabilityQuotas;
use super::manifest::{SandboxCapability, SandboxManifest};
use quota::QuotaLedger;

pub use audit::{
    ActivitySummary, CapabilityEvent, CapabilityOutcome, DroppedCalls, MAX_LOG_EVENTS,
    PluginActivityLog,
};
// Aliased for use *inside* this module only, so the `pub use` above can carry
// the real names without shadowing itself.
use audit::{CapabilityEvent as Event, CapabilityOutcome as Outcome};
pub use db::{MemoryPluginStore, PluginStore, QueryPage, Scope, StoreError};
pub use jobs::MemoryJobSink;
pub use jobs::{JobSink, PluginJob};
pub use kv::{CacheKvStore, KvStore, MemoryKvStore};
pub use outbound::{
    ALLOWED_OUTBOUND_REQUEST_HEADERS, ALLOWED_OUTBOUND_RESPONSE_HEADERS, MAX_OUTBOUND_HEADERS,
    MAX_RESPONSE_HEADER_BYTES, OutboundHttp, OutboundRequest, OutboundResponse, RecordingHttp,
};
pub use quota::CapabilityRateLimiter;
pub use render::{ALLOWED_ATTRIBUTES, ALLOWED_TAGS, FragmentNode, RenderError};

/// The most columns one plugin row may carry.
///
/// A row crosses the wire as a map the guest wrote, is held in host memory, and
/// (for the DB capability) becomes a statement's column list. Every one of those
/// is work an untrusted map sizes, so the count is bounded before any of it
/// happens.
pub const MAX_ROW_COLUMNS: usize = 32;

/// Longest accepted text in one row value or KV key, in bytes.
pub const MAX_VALUE_TEXT_BYTES: usize = 64 * 1024;

/// Longest accepted logical KV key, in bytes.
///
/// The key is the one identifier a guest chooses freely — tables, hosts, job
/// types and slots all have to appear in the manifest — so it is the one that
/// needs a length of its own rather than inheriting a grant's.
pub const MAX_KV_KEY_BYTES: usize = 512;

/// Most bytes one row may carry across all of its columns.
///
/// [`MAX_VALUE_TEXT_BYTES`] bounds one column, and 32 of them bounds a row at
/// 2 MiB — twice what a whole reply may be. A row that cannot be handed back is
/// a row that must not be accepted, so the ceiling that lets a `db-get` answer
/// is applied where the row goes *in*.
pub const MAX_ROW_BYTES: usize = 256 * 1024;

/// Most bytes the rows in one `db-get` or `db-query` answer may carry.
///
/// Rows are the one result whose size the *guest* chooses: a `db_rows` quota of
/// 500 against [`MAX_ROW_BYTES`] is 125 MiB the host would materialise, encode
/// and hold, per request, on a quota the plugin's own manifest sets. So the
/// budget travels into the store (see [`PluginStore::query`]) rather than being
/// checked once the whole answer exists — by which point the allocation has
/// already happened.
///
/// Comfortably under the host's queued-reply ceiling, so a reply that fits this
/// always fits the queue and a legal read never fails a later call.
///
/// [`PluginStore::query`]: db::PluginStore::query
pub const MAX_RESULT_BYTES: usize = 512 * 1024;

/// Longest accepted plugin row id, in bytes.
pub const MAX_ROW_ID_BYTES: usize = 128;

/// The longest job id a [`JobSink`] may assign.
///
/// The id goes back to the guest after the job is queued, so it is bounded where
/// it enters the reply.
pub const MAX_JOB_ID_BYTES: usize = 128;

// ── Values ───────────────────────────────────────────────────────────────

/// One scalar in a plugin row or job payload.
///
/// Deliberately **not** `serde_json::Value`. Arbitrary JSON is a recursive
/// structure an untrusted guest chooses the depth of, and every consumer of it
/// — the row encoder, the store, the job payload — would need its own depth
/// bound to be safe. A closed set of scalars has no depth to bound, so the only
/// ceilings left are a column count and a byte length, both checked once here.
///
/// The `untagged` representation is what a plugin author would write by hand:
/// `{"sku":"A-1","qty":3,"active":true}` rather than
/// `{"sku":{"text":"A-1"}}`. The variants are ordered so the match is
/// unambiguous — a JSON integer is `Int` because `Bool` cannot hold it and
/// `Int` is tried before `Float`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum PluginValue {
    /// JSON `null`.
    Null,
    /// JSON `true` / `false`.
    Bool(bool),
    /// A JSON integer.
    Int(i64),
    /// A JSON number that is not an integer.
    Float(f64),
    /// A JSON string.
    Text(String),
}

thread_local! {
    /// Nodes still allowed in the fragment tree currently being deserialized.
    ///
    /// A thread-local rather than a parameter because `deserialize_with` takes
    /// only a `Deserializer`, and the budget has to be shared by a vector and
    /// every vector nested inside it. Each request runs on its own blocking
    /// thread, so there is no sharing between plugins, and the guard below
    /// restores the previous value on the way out — including on the error
    /// path, which is the common one here.
    static FRAGMENT_NODE_BUDGET: std::cell::Cell<usize> =
        const { std::cell::Cell::new(usize::MAX) };
}

/// Reserve a whole-tree node budget for as long as this value lives.
pub struct FragmentNodeBudget(usize);

impl FragmentNodeBudget {
    /// Start a fragment parse with `limit` nodes for the entire tree.
    #[must_use]
    pub fn reserve(limit: usize) -> Self {
        Self(FRAGMENT_NODE_BUDGET.with(|budget| budget.replace(limit)))
    }
}

impl Drop for FragmentNodeBudget {
    fn drop(&mut self) {
        FRAGMENT_NODE_BUDGET.with(|budget| budget.set(self.0));
    }
}

/// Deserialize the *root* of a fragment tree, reserving a budget for all of it.
///
/// # Errors
///
/// When the tree carries more than `LIMIT` nodes in total.
pub fn bounded_tree<'de, D, const LIMIT: usize>(
    deserializer: D,
) -> Result<Vec<render::FragmentNode>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let _budget = FragmentNodeBudget::reserve(LIMIT);
    bounded_subtree(deserializer)
}

/// Deserialize nodes against the budget the root reserved.
///
/// Per-vector ceilings were not enough, and this is the third pass over the
/// same ground: bounding each `children` vector at `MAX_NODES` still admits
/// `MAX_NODES` of them nested inside each other, so a frame within the stdout
/// budget deserialized hundreds of thousands of nodes and was refused only
/// afterwards, by `render`, on the aggregate. The count has to be shared by the
/// whole tree or it is not a bound on the tree at all.
///
/// # Errors
///
/// When the tree's shared budget runs out.
pub fn bounded_subtree<'de, D>(deserializer: D) -> Result<Vec<render::FragmentNode>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    use serde::de::{Error as _, SeqAccess, Visitor};

    struct Bounded;

    impl<'de> Visitor<'de> for Bounded {
        type Value = Vec<render::FragmentNode>;

        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            write!(
                formatter,
                "at most {} nodes in the whole fragment",
                render::MAX_NODES
            )
        }

        fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Self::Value, A::Error> {
            let mut out = Vec::new();
            // Read, then charge -- not the other way round. Charging first
            // meant the probe that discovers the *end* of the sequence was
            // itself charged, so a vector holding exactly the budget was
            // refused for being one over it. An element's own children are
            // charged as they are read, inside this same call, so the tree is
            // still bounded while it is built rather than after.
            while let Some(node) = seq.next_element()? {
                let spent = FRAGMENT_NODE_BUDGET.with(|budget| {
                    let left = budget.get();
                    if left == 0 {
                        return false;
                    }
                    budget.set(left.saturating_sub(1));
                    true
                });
                if !spent {
                    return Err(A::Error::custom(format!(
                        "more than {} nodes in the fragment",
                        render::MAX_NODES
                    )));
                }
                out.push(node);
            }
            Ok(out)
        }
    }

    deserializer.deserialize_seq(Bounded)
}

/// Deserialize a `Vec` that refuses to grow past `LIMIT` while it is being read.
///
/// A structural ceiling checked *after* `serde_json` has built the collection
/// bounds what is rendered or dispatched, never what is allocated. One frame of
/// a few hundred kilobytes can carry hundreds of thousands of `["",""]` pairs,
/// each of which becomes two `String`s — tens of megabytes of host memory,
/// outside the guest's own limiter and multiplied by every concurrent request,
/// spent to produce a value that is then refused for having too many elements.
///
/// The limit is applied by the visitor, so the refusal happens at element
/// `LIMIT + 1` and nothing larger is ever held.
///
/// # Errors
///
/// The deserializer's own error when an element cannot be read, and a custom
/// one naming `LIMIT` when the sequence carries more than that many.
pub fn bounded_vec<'de, D, T, const LIMIT: usize>(deserializer: D) -> Result<Vec<T>, D::Error>
where
    D: serde::Deserializer<'de>,
    T: serde::Deserialize<'de>,
{
    use serde::de::{Error as _, SeqAccess, Visitor};

    struct Bounded<T, const LIMIT: usize>(std::marker::PhantomData<T>);

    impl<'de, T, const LIMIT: usize> Visitor<'de> for Bounded<T, LIMIT>
    where
        T: serde::Deserialize<'de>,
    {
        type Value = Vec<T>;

        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            write!(formatter, "at most {LIMIT} elements")
        }

        fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Self::Value, A::Error> {
            // `size_hint` is a hint from the input and not to be trusted with an
            // allocation, so it is clamped to what will be accepted anyway.
            let mut out = Vec::with_capacity(seq.size_hint().unwrap_or(0).min(LIMIT));
            while let Some(item) = seq.next_element()? {
                if out.len() >= LIMIT {
                    return Err(A::Error::custom(format!("more than {LIMIT} elements")));
                }
                out.push(item);
            }
            Ok(out)
        }
    }

    deserializer.deserialize_seq(Bounded::<T, LIMIT>(std::marker::PhantomData))
}

/// Deserialize a [`PluginRow`] that refuses to grow past [`MAX_ROW_COLUMNS`]
/// while it is being read.
///
/// [`bounded_vec`] for maps, and needed for the same reason: `check_row` applies
/// the column ceiling to a map `serde_json` has already built, so a frame of
/// hundreds of thousands of one-character columns is materialised in full and
/// then refused for having too many. Worse than the sequence case, because
/// `dispatch` deserializes before it checks the capability — so an *ungranted*
/// call can spend that memory.
///
/// # Errors
///
/// The deserializer's own error when an entry cannot be read, and a custom one
/// naming the ceiling when the map carries more entries than that.
pub fn bounded_row<'de, D>(deserializer: D) -> Result<PluginRow, D::Error>
where
    D: serde::Deserializer<'de>,
{
    use serde::de::{Error as _, MapAccess, Visitor};

    struct Bounded;

    impl<'de> Visitor<'de> for Bounded {
        type Value = PluginRow;

        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            write!(formatter, "at most {} columns", MAX_ROW_COLUMNS + 1)
        }

        fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Self::Value, A::Error> {
            // One over the column ceiling, deliberately. `db-get` hands the row
            // back with the host's `row_id` on it, and the documented
            // read-modify-write is to change a field and send that map to
            // `db-update` — so a row stored at exactly `MAX_ROW_COLUMNS`
            // arrives here with 33 entries. Refusing it *here* would fail the
            // whole frame as malformed before `validated_row` could strip the
            // column the host itself added, which is the bug this subsystem
            // already fixed once at the dispatch layer and this parse bound
            // reintroduced one layer earlier.
            //
            // The real limit is still enforced, after stripping, by
            // `validated_row`. This is only a bound on what is *allocated*, and
            // thirty-three entries bounds it exactly as well as thirty-two.
            const PARSE_CEILING: usize = MAX_ROW_COLUMNS + 1;

            let mut out = PluginRow::new();
            while let Some((column, value)) = map.next_entry::<String, PluginValue>()? {
                // Counted on the map, not on entries read: a frame repeating one
                // column name collapses to a single entry, and refusing it for
                // its *reads* would reject a legal row written wastefully.
                if out.len() >= PARSE_CEILING && !out.contains_key(&column) {
                    return Err(A::Error::custom(format!(
                        "more than {PARSE_CEILING} columns"
                    )));
                }
                out.insert(column, value);
            }
            Ok(out)
        }
    }

    deserializer.deserialize_map(Bounded)
}

/// The bytes `text` occupies once JSON-escaped, excluding its framing quotes.
///
/// Not `text.len()`. JSON escaping is not a constant factor: a NUL becomes the
/// six bytes `\u0000`, so 256 KiB of them serialize past 1.5 MiB. Every ceiling
/// in this module exists to stop the host building a reply it cannot send, and
/// a ceiling measured in *stored* bytes does not do that — it passes a value
/// that then overruns the reply queue and fails the request, after whatever
/// side effect produced it has already happened.
///
/// The two framing quotes are deliberately *not* counted. They are per-value
/// encoding overhead, already covered by the per-entry constants the stores
/// add, and counting them here would quietly redefine `kv_value_bytes` and
/// every sibling ceiling as two bytes less than the number the operator wrote —
/// a value of exactly the declared size would be refused for carrying its own
/// punctuation.
///
/// Counted rather than serialized, so asking the size costs no allocation.
#[must_use]
pub fn encoded_text_len(text: &str) -> usize {
    text.chars()
        .map(|ch| match ch {
            // The two-byte escapes JSON names.
            '"' | '\\' | '\n' | '\r' | '\t' | '\u{8}' | '\u{c}' => 2,
            // Every other control character takes the `\u00XX` long form.
            other if other.is_control() => 6,
            other => other.len_utf8(),
        })
        .fold(0, usize::saturating_add)
}

/// How many characters `value` occupies in decimal, sign included.
///
/// Counted rather than formatted: `weight` runs per value on the write path.
const fn decimal_len(value: i64) -> usize {
    // `i64::MIN` has no positive counterpart, so it cannot be negated to be
    // counted. It is also the longest, which makes the special case the answer.
    if value == i64::MIN {
        return 20;
    }
    let negative = value < 0;
    let mut left = value.unsigned_abs();
    let mut digits = 1_usize;
    while left >= 10 {
        left /= 10;
        digits = digits.saturating_add(1);
    }
    if negative {
        digits.saturating_add(1)
    } else {
        digits
    }
}

impl PluginValue {
    /// The bytes this value costs, measured as it will be *encoded*.
    ///
    /// See [`encoded_text_len`]: the quotas this feeds bound a reply the host
    /// has to serialize, so the stored length is the wrong number.
    #[must_use]
    pub fn weight(&self) -> usize {
        match self {
            Self::Text(text) => encoded_text_len(text),
            // Measured, not assumed. A flat eight bytes for every scalar was
            // wrong in the direction that matters: `i64::MIN` encodes as
            // twenty, so a `kv_value_bytes` of eight admitted a value the
            // operator's ceiling forbade — on the way in *and* on the way back
            // out, since reads are checked against the same number.
            // `null` and `true` are both four characters; `false` is five.
            Self::Null | Self::Bool(true) => 4,
            Self::Bool(false) => 5,
            Self::Int(value) => decimal_len(*value),
            // A bound rather than a measurement, and deliberately: the shortest
            // round-trip form of an `f64` is at most twenty-four characters
            // (`-2.2250738585072014e-308`), so this can never *under*-count,
            // which is the only direction a ceiling may not be wrong in.
            // Formatting one to find out would allocate on a path that runs per
            // value.
            Self::Float(_) => 24,
        }
    }
}

impl fmt::Display for PluginValue {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Null => f.write_str("null"),
            Self::Bool(value) => write!(f, "{value}"),
            Self::Int(value) => write!(f, "{value}"),
            Self::Float(value) => write!(f, "{value}"),
            Self::Text(value) => f.write_str(value),
        }
    }
}

/// One row: column name to scalar, ordered so encodings are deterministic.
pub type PluginRow = BTreeMap<String, PluginValue>;

/// Check one row against the bounds every backend inherits.
///
/// Checked in the host rather than in each backend, so a store implementation
/// an embedder wrote cannot be the place the bound is missing.
///
/// # Errors
///
/// Names the first rule the row broke.
pub fn check_row(row: &PluginRow) -> Result<(), String> {
    check_row_without(row, None)
}

/// As [`check_row`], ignoring `skip` if the row carries it.
///
/// Exists so a caller that is going to strip a column can check the row it will
/// *build* without building it first. Stripping by cloning and removing meant a
/// row over the ceiling was copied in full before the ceiling refused it, and a
/// legal one was copied twice — once to strip, once to return.
///
/// # Errors
///
/// One line naming the ceiling the row missed.
pub fn check_row_without(row: &PluginRow, skip: Option<&str>) -> Result<(), String> {
    let kept = || {
        row.iter()
            .filter(move |(column, _)| Some(column.as_str()) != skip)
    };
    let columns = kept().count();
    if columns > MAX_ROW_COLUMNS {
        return Err(format!(
            "a row may carry at most {MAX_ROW_COLUMNS} columns; this one carries {columns}"
        ));
    }
    let weight = kept().fold(0, |sum: usize, (column, value)| {
        sum.saturating_add(column.len())
            .saturating_add(value.weight())
    });
    if weight > MAX_ROW_BYTES {
        return Err(format!(
            "a row may carry at most {MAX_ROW_BYTES} bytes across its columns; this one carries \
             {weight}"
        ));
    }
    for (column, value) in kept() {
        if !super::grants::is_grantable_ident(column) {
            return Err(format!(
                "column name {:?} is not a lower-case `[a-z][a-z0-9_]*` identifier; the host \
                 derives a physical column name from it",
                super::manifest::rejected(column)
            ));
        }
        if value.weight() > MAX_VALUE_TEXT_BYTES {
            return Err(format!(
                "column {column:?} carries {} bytes, over the {MAX_VALUE_TEXT_BYTES}-byte ceiling",
                value.weight()
            ));
        }
    }
    Ok(())
}

/// What one row costs in host memory: its values, plus its column names.
///
/// Column names count because they are repeated in every row of an answer and
/// in the JSON of every reply, so a row of 32 long names is not free.
#[must_use]
pub fn row_weight(row: &PluginRow) -> usize {
    row.iter().fold(0, |sum, (column, value)| {
        sum.saturating_add(column.len())
            .saturating_add(value.weight())
    })
}

/// Bytes of JSON syntax around a row: quotes, colons and commas per column,
/// and the braces and `row_id` key.
pub(super) fn row_framing(row: &PluginRow) -> usize {
    row.len().saturating_mul(8).saturating_add(16)
}

/// Keep rows while they fit [`MAX_RESULT_BYTES`], reporting whether any were
/// left behind.
///
/// The second line of defence, not the first: [`PluginStore::query`] is given
/// the same budget so a well-behaved store never builds the rows this would
/// discard. It runs anyway, because the store is an embedder's trait
/// implementation and "the host does not hold more than this" cannot rest on
/// someone else's loop.
///
/// A row the host would have accepted always survives: [`MAX_ROW_BYTES`] is
/// under this ceiling, so a `db-get` of a legally-stored row is never answered
/// with nothing. A row *larger* than the whole result budget is dropped rather
/// than exempted — it can only have come from a store that returned what the
/// host would never have stored, and passing it through is exactly the
/// oversized reply this exists to refuse.
///
/// [`PluginStore::query`]: db::PluginStore::query
#[must_use]
pub fn bounded_rows(rows: Vec<PluginRow>) -> (Vec<PluginRow>, bool) {
    let mut kept = Vec::with_capacity(rows.len());
    let mut total = 0_usize;
    let mut truncated = false;
    for row in rows {
        // JSON syntax counts too. Else many narrow rows exceed the reply room.
        let weight = row_weight(&row).saturating_add(row_framing(&row));
        // No "but keep the first one" exemption. Letting an empty `kept` skip
        // the ceiling meant one corrupt or oversized row from an embedder's
        // store was serialized in full — and a single row can be arbitrarily
        // large when it did not come through `check_row`.
        if total.saturating_add(weight) > MAX_RESULT_BYTES {
            truncated = true;
            break;
        }
        total = total.saturating_add(weight);
        kept.push(row);
    }
    (kept, truncated)
}

// ── The calls ────────────────────────────────────────────────────────────

/// One thing a guest asks the host to do on its behalf.
///
/// Internally tagged by `call`, so a frame reads as
/// `{"op":"call","call":"kv-get","id":1,"key":"cart"}` — one flat object a
/// plugin author can write without a code generator, which is the same reason
/// the request frame is what it is.
///
/// There is no variant carrying SQL, a URL path on the host, a tenant, or a
/// physical table name, and that absence is the design: a call cannot ask for
/// something out of scope because the wire has nowhere to put it.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "call", rename_all = "kebab-case", deny_unknown_fields)]
#[non_exhaustive]
pub enum CapabilityCall {
    /// Read one key from the plugin's per-tenant namespace.
    KvGet {
        /// Correlation id the result echoes.
        id: u64,
        /// The logical key.
        key: String,
    },
    /// Write one key.
    KvSet {
        /// Correlation id.
        id: u64,
        /// The logical key.
        key: String,
        /// The value.
        value: PluginValue,
    },
    /// Delete one key.
    KvDelete {
        /// Correlation id.
        id: u64,
        /// The logical key.
        key: String,
    },
    /// Call a declared upstream.
    HttpFetch {
        /// Correlation id.
        id: u64,
        /// HTTP method.
        method: String,
        /// Absolute URL. Its host must appear in `[grants].hosts`.
        url: String,
        /// Request headers.
        ///
        /// Bounded while it is read, not after the allow-list check; see
        /// [`bounded_vec`].
        #[serde(
            default,
            deserialize_with = "bounded_vec::<_, _, MAX_OUTBOUND_HEADERS>"
        )]
        headers: Vec<(String, String)>,
        /// Request body, as text.
        #[serde(default)]
        body: String,
    },
    /// Insert one row into a declared plugin table.
    DbInsert {
        /// Correlation id.
        id: u64,
        /// Logical table name, which must appear in `[grants].tables`.
        table: String,
        /// The row.
        #[serde(deserialize_with = "bounded_row")]
        row: PluginRow,
    },
    /// Read one row by id.
    DbGet {
        /// Correlation id.
        id: u64,
        /// Logical table name.
        table: String,
        /// The row id, as returned by `db-insert`.
        row_id: String,
    },
    /// Read rows matching an equality filter.
    DbQuery {
        /// Correlation id.
        id: u64,
        /// Logical table name.
        table: String,
        /// Column equality filter; an empty map matches every row the plugin
        /// and tenant own.
        #[serde(default, deserialize_with = "bounded_row")]
        filter: PluginRow,
        /// Largest number of rows to return, capped by the `db_rows` quota.
        #[serde(default)]
        limit: u32,
        /// Resume after this `row_id`, for reading past one page.
        ///
        /// A page can end early on the `db_rows` quota or on
        /// [`MAX_RESULT_BYTES`], and `truncated` tells the guest that happened —
        /// but without a place to say "carry on from here", the same filter
        /// would return the same prefix forever and the rows behind it would be
        /// unreachable. Rows come back in ascending `row_id` order (see
        /// [`PluginStore::query`](db::PluginStore::query)), so the last
        /// `row_id` of one page is the `after` of the next.
        #[serde(default)]
        after: Option<String>,
    },
    /// Replace one row by id.
    DbUpdate {
        /// Correlation id.
        id: u64,
        /// Logical table name.
        table: String,
        /// The row id.
        row_id: String,
        /// The replacement row.
        #[serde(deserialize_with = "bounded_row")]
        row: PluginRow,
    },
    /// Delete one row by id.
    DbDelete {
        /// Correlation id.
        id: u64,
        /// Logical table name.
        table: String,
        /// The row id.
        row_id: String,
    },
    /// Enqueue one declared job type.
    JobEnqueue {
        /// Correlation id.
        id: u64,
        /// The job type, which must appear in `[grants].job_types`.
        job_type: String,
        /// The job's arguments.
        #[serde(default, deserialize_with = "bounded_row")]
        payload: PluginRow,
    },
}

impl CapabilityCall {
    /// The correlation id the result echoes.
    #[must_use]
    pub const fn id(&self) -> u64 {
        match self {
            Self::KvGet { id, .. }
            | Self::KvSet { id, .. }
            | Self::KvDelete { id, .. }
            | Self::HttpFetch { id, .. }
            | Self::DbInsert { id, .. }
            | Self::DbGet { id, .. }
            | Self::DbQuery { id, .. }
            | Self::DbUpdate { id, .. }
            | Self::DbDelete { id, .. }
            | Self::JobEnqueue { id, .. } => *id,
        }
    }

    /// The capability this call needs.
    #[must_use]
    pub const fn capability(&self) -> SandboxCapability {
        match self {
            Self::KvGet { .. } | Self::KvSet { .. } | Self::KvDelete { .. } => {
                SandboxCapability::Kv
            }
            Self::HttpFetch { .. } => SandboxCapability::HttpOutbound,
            Self::DbInsert { .. }
            | Self::DbGet { .. }
            | Self::DbQuery { .. }
            | Self::DbUpdate { .. }
            | Self::DbDelete { .. } => SandboxCapability::Db,
            Self::JobEnqueue { .. } => SandboxCapability::Jobs,
        }
    }

    /// The logical thing this call names, before any check has run.
    ///
    /// Distinct from the *validated* target: a call refused for naming a host
    /// the manifest never granted still names one, and an operator asking "what
    /// did this plugin reach for" is asking exactly about those. Recording only
    /// what passed would make the audit surface answer the easy half of the
    /// question.
    ///
    /// The URL case reports the **host**, not the URL: a URL is a guest-chosen
    /// string with a query string on it, and the audit ledger is a bounded
    /// buffer that is not the place to keep one.
    #[must_use]
    pub fn target(&self) -> String {
        match self {
            Self::KvGet { key, .. } | Self::KvSet { key, .. } | Self::KvDelete { key, .. } => {
                key.clone()
            }
            // The *authority* it reached for, even when `host_of` refuses the
            // URL — because the refusals are the interesting ones. Recording a
            // placeholder would put every userinfo, backslash and scheme-
            // confusion attempt under one label, so `attacker.test` in
            // `https://api.example.com@attacker.test/` would never appear in the
            // one surface built to show what a plugin reached for.
            Self::HttpFetch { url, .. } => {
                outbound::host_of(url).unwrap_or_else(|| outbound::attempted_authority(url))
            }
            Self::DbInsert { table, .. }
            | Self::DbGet { table, .. }
            | Self::DbQuery { table, .. }
            | Self::DbUpdate { table, .. }
            | Self::DbDelete { table, .. } => table.clone(),
            Self::JobEnqueue { job_type, .. } => job_type.clone(),
        }
    }

    /// Bytes a reply to this call may take, at its worst.
    ///
    /// A read and an outbound call can have a large reply. A write answers with
    /// `done` or an id. Headers count at the JSON worst case of six bytes per
    /// byte. The host refuses a call that has no room, so the guest gets a
    /// denial it can read, not a failed request.
    #[must_use]
    pub const fn reply_reserve(&self) -> usize {
        self.reply_reserve_for(MAX_RESULT_BYTES, 500)
    }

    /// As [`reply_reserve`](Self::reply_reserve), with the plugin's own
    /// `kv_value_bytes` for a KV read and `db_rows` for a query.
    ///
    /// A DB reply adds JSON framing to the row weight: quotes, colons and
    /// commas for each column of each row.
    #[must_use]
    pub const fn reply_reserve_for(&self, kv_value_bytes: usize, db_rows: usize) -> usize {
        /// Framing bytes for one row at the column limit, with its `row_id`.
        const ROW_FRAMING: usize = (MAX_ROW_COLUMNS + 1) * 8 + 16;
        match self {
            Self::KvGet { .. } => kv_value_bytes.saturating_add(4096),
            Self::DbGet { .. } => MAX_RESULT_BYTES + ROW_FRAMING + 4096,
            Self::DbQuery { limit, .. } => {
                // The page the call can return: its own limit, within the quota.
                let asked = *limit as usize;
                let rows = if asked == 0 || asked > db_rows {
                    db_rows
                } else {
                    asked
                };
                let reserve = MAX_RESULT_BYTES
                    .saturating_add(rows.saturating_mul(ROW_FRAMING))
                    .saturating_add(4096);
                // Never more than the reply queue holds, or no query could run.
                if reserve > super::host::MAX_QUEUED_REPLY_BYTES {
                    super::host::MAX_QUEUED_REPLY_BYTES
                } else {
                    reserve
                }
            }
            Self::HttpFetch { .. } => {
                MAX_RESULT_BYTES + outbound::MAX_RESPONSE_HEADER_BYTES * 6 + 4096
            }
            _ => 4096,
        }
    }

    /// [`target`](Self::target), already cut to what the ledger keeps.
    ///
    /// Used before any check has run, so it never copies more than
    /// [`MAX_TARGET_CHARS`] characters of a guest string. A URL is cut first, to
    /// keep the host parse bounded too.
    #[must_use]
    pub fn target_excerpt(&self) -> String {
        /// Bytes of URL parsed as a whole; a longer URL is scanned in place.
        const URL_PREFIX_CHARS: usize = 4096;
        match self {
            Self::KvGet { key, .. } | Self::KvSet { key, .. } | Self::KvDelete { key, .. } => {
                bounded_target(key)
            }
            Self::HttpFetch { url, .. } if url.len() <= URL_PREFIX_CHARS => bounded_target(
                &outbound::host_of(url).unwrap_or_else(|| outbound::attempted_authority(url)),
            ),
            // A long URL: scan for the authority in place, copy nothing, and
            // keep the text after the last `@` as a client would.
            Self::HttpFetch { url, .. } => {
                let rest = url.split_once("://").map_or(url.as_str(), |(_, rest)| rest);
                let rest = rest.trim_start_matches('/');
                let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
                let host = authority.rsplit('@').next().unwrap_or(authority);
                bounded_target(host).to_ascii_lowercase()
            }
            Self::DbInsert { table, .. }
            | Self::DbGet { table, .. }
            | Self::DbQuery { table, .. }
            | Self::DbUpdate { table, .. }
            | Self::DbDelete { table, .. } => bounded_target(table),
            Self::JobEnqueue { job_type, .. } => bounded_target(job_type),
        }
    }

    /// The operation name the audit ledger records.
    #[must_use]
    pub const fn operation(&self) -> &'static str {
        match self {
            Self::KvGet { .. } => "kv-get",
            Self::KvSet { .. } => "kv-set",
            Self::KvDelete { .. } => "kv-delete",
            Self::HttpFetch { .. } => "http-fetch",
            Self::DbInsert { .. } => "db-insert",
            Self::DbGet { .. } => "db-get",
            Self::DbQuery { .. } => "db-query",
            Self::DbUpdate { .. } => "db-update",
            Self::DbDelete { .. } => "db-delete",
            Self::JobEnqueue { .. } => "job-enqueue",
        }
    }

    /// Whether this call can change state outside the guest.
    ///
    /// A failed request that made such a call can make it again on retry. An
    /// outbound `GET` or `HEAD` is a read. An unknown method counts as a write.
    #[must_use]
    pub fn commits(&self) -> bool {
        match self {
            Self::KvGet { .. } | Self::DbGet { .. } | Self::DbQuery { .. } => false,
            Self::HttpFetch { method, .. } => {
                !(method.eq_ignore_ascii_case("GET") || method.eq_ignore_ascii_case("HEAD"))
            }
            Self::KvSet { .. }
            | Self::KvDelete { .. }
            | Self::DbInsert { .. }
            | Self::DbUpdate { .. }
            | Self::DbDelete { .. }
            | Self::JobEnqueue { .. } => true,
        }
    }
}

// ── The results ──────────────────────────────────────────────────────────

/// Why a call was refused.
///
/// A single machine-readable word, because a guest that can tell "you may not"
/// from "not this time" can degrade correctly, and an author reading a denial
/// needs to know which rule to fix.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
#[non_exhaustive]
pub enum DenialReason {
    /// The manifest does not grant the capability this call needs.
    CapabilityNotGranted,
    /// The capability is granted, but `[grants]` does not name this target.
    NotInGrant,
    /// A per-request quota is spent.
    QuotaExceeded,
    /// The call itself is malformed — an over-long key, an unparseable URL.
    Malformed,
    /// The host has no backend wired for this capability.
    Unavailable,
    /// The backend was reached and refused or failed.
    BackendError,
    /// An upstream answered, and its answer was over a byte ceiling.
    ///
    /// Distinct from [`QuotaExceeded`](Self::QuotaExceeded) because the two mean
    /// opposite things to an operator: a quota denial means nothing left the
    /// host, and this means the call *was made* and the answer was discarded.
    /// The audit surface counts the host as called.
    ResponseTooLarge,
    /// The call ran, and the host refused the answer the backend gave.
    ///
    /// The write or enqueue may have happened, so a guest MUST NOT retry. The
    /// audit surface counts the target as touched.
    ResultRejected,
}

impl DenialReason {
    /// The wire spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::CapabilityNotGranted => "capability-not-granted",
            Self::NotInGrant => "not-in-grant",
            Self::QuotaExceeded => "quota-exceeded",
            Self::Malformed => "malformed",
            Self::Unavailable => "unavailable",
            Self::BackendError => "backend-error",
            Self::ResponseTooLarge => "response-too-large",
            Self::ResultRejected => "result-rejected",
        }
    }
}

impl fmt::Display for DenialReason {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

/// What the host hands back for one call.
///
/// `#[must_use]` because dropping it drops the guest's *answer*: the frame the
/// host owes it in return for the call it made.
#[must_use]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "kebab-case")]
#[non_exhaustive]
pub enum CallResult {
    /// The call succeeded.
    Ok {
        /// Correlation id, echoed.
        id: u64,
        /// The answer.
        value: CallValue,
    },
    /// The call was refused.
    Denied {
        /// Correlation id, echoed.
        id: u64,
        /// Which rule refused it.
        reason: DenialReason,
        /// One line for the author, already bounded and escaped.
        detail: String,
    },
}

/// Characters of a denial detail that reach the guest and the log.
///
/// The host's own sentences are all far shorter. What this bounds is the half
/// it does not write: a `StoreError::Backend`, a `JobSink` error or an
/// `OutboundHttp` failure is a string from someone else's driver, and a driver
/// that echoes the query, the row or the upstream's body makes it as long as
/// whoever influenced that input cared to. Generous next to any real message
/// and far under the reply queue, so a denial can never be the thing that fails
/// the request it was explaining.
pub const MAX_DETAIL_CHARS: usize = 512;

impl CallResult {
    /// Build a denial.
    ///
    /// Bounds and neutralises `detail` here rather than at each call site,
    /// because the field's contract says it is bounded and escaped and the
    /// interesting inputs are the ones the host did not write: a backend's
    /// error string is an embedder's driver talking, and it may carry as much
    /// attacker-influenced text as that driver chose to include. Serialized in
    /// full before the reply-queue ceiling is checked, an unbounded one would
    /// turn an explanation into the failure it was explaining.
    pub fn denied(id: u64, reason: DenialReason, detail: impl Into<String>) -> Self {
        let detail = detail.into();
        let mut chars = detail.chars();
        let kept: String = chars.by_ref().take(MAX_DETAIL_CHARS).collect();
        let kept = if chars.next().is_some() {
            format!("{kept}…")
        } else {
            kept
        };
        Self::Denied {
            id,
            reason,
            // Escaped as well as bounded: this reaches a `tracing` line and the
            // rendered operator summary, which a terminal interprets.
            detail: super::manifest::rejected(&kept),
        }
    }

    /// The correlation id.
    #[must_use]
    pub const fn id(&self) -> u64 {
        match self {
            Self::Ok { id, .. } | Self::Denied { id, .. } => *id,
        }
    }

    /// The denial reason, if this is one.
    #[must_use]
    pub const fn denial(&self) -> Option<DenialReason> {
        match self {
            Self::Ok { .. } => None,
            Self::Denied { reason, .. } => Some(*reason),
        }
    }
}

/// The answer to a successful call.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
#[non_exhaustive]
pub enum CallValue {
    /// Nothing to say — a successful write or delete.
    Done,
    /// One optional scalar, from `kv-get`.
    Value {
        /// The value, or `null` for a miss. `found` distinguishes a stored
        /// `null` from a missing key, which a `null` alone could not.
        value: PluginValue,
        /// Whether the key existed.
        found: bool,
    },
    /// A row id, from `db-insert`.
    RowId {
        /// The id the store assigned.
        row_id: String,
    },
    /// Rows, from `db-get` and `db-query`.
    Rows {
        /// The rows, each carrying its `row_id` alongside its columns.
        rows: Vec<PluginRow>,
        /// Whether [`MAX_RESULT_BYTES`] stopped the answer short.
        ///
        /// On the wire so a guest can tell "that is all of them" from "that is
        /// all that fits". Without it a plugin paging through its own table
        /// would read a short answer as the end of the table and stop, which is
        /// silent data loss in the plugin rather than in the host.
        truncated: bool,
    },
    /// An HTTP response, from `http-fetch`.
    Http {
        /// The upstream status code.
        status: u16,
        /// Response headers the host allowed through.
        headers: Vec<(String, String)>,
        /// Response body, as text.
        body: String,
    },
    /// A job id, from `job-enqueue`.
    JobId {
        /// The id the queue assigned.
        job_id: String,
    },
}

// ── The services ─────────────────────────────────────────────────────────

/// The backends a host will honour a capability against.
///
/// Deliberately **not** `#[non_exhaustive]`, unlike the types in this module
/// that the host produces and a caller only reads: this one is built by the
/// embedder, and `CapabilityServices { kv: …, ..CapabilityServices::none() }` is
/// how. Marking it would make that spelling impossible outside this crate,
/// which is the only spelling there is.
///
/// Every field is optional, and a missing one is not a hole: a call whose
/// backend is absent is denied [`DenialReason::Unavailable`] and recorded, the
/// same as any other refusal. That is what lets `SandboxHost::run` — which has
/// no application state at all — keep working unchanged, and what lets a test
/// exercise one capability without standing up the other four.
#[derive(Clone, Default)]
pub struct CapabilityServices {
    /// The tenant every scoped capability binds to.
    ///
    /// `None` is the single-tenant application: keys and rows land in a
    /// namespace of their own rather than in some other tenant's.
    pub tenant: Option<String>,
    /// Backing store for `kv`.
    pub kv: Option<Arc<dyn KvStore>>,
    /// Client for `http-outbound`.
    pub http: Option<Arc<dyn OutboundHttp>>,
    /// Store for `db`.
    pub db: Option<Arc<dyn PluginStore>>,
    /// Queue for `jobs`.
    pub jobs: Option<Arc<dyn JobSink>>,
    /// The plugin's own calls-per-second ceiling, shared across its requests.
    ///
    /// `None` leaves only the per-request counters, which is the right default
    /// for a direct `SandboxHost::run` caller: a rate ceiling with no shared
    /// state is a ceiling that resets every call, which is worse than saying it
    /// is not there. `SandboxedPlugin` builds one per plugin.
    pub rate: Option<Arc<quota::CapabilityRateLimiter>>,
}

impl fmt::Debug for CapabilityServices {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CapabilityServices")
            .field("tenant", &self.tenant)
            .field("kv", &self.kv.is_some())
            .field("http", &self.http.is_some())
            .field("db", &self.db.is_some())
            .field("jobs", &self.jobs.is_some())
            .field("rate", &self.rate.is_some())
            .finish()
    }
}

impl CapabilityServices {
    /// No backends at all — every capability call is `unavailable`.
    #[must_use]
    pub fn none() -> Self {
        Self::default()
    }

    /// The same services, bound to `tenant`.
    #[must_use]
    pub fn for_tenant(mut self, tenant: impl Into<String>) -> Self {
        self.tenant = Some(tenant.into());
        self
    }
}

// ── The runtime ──────────────────────────────────────────────────────────

/// The tenant namespace segment used when the application is single-tenant.
///
/// A literal rather than an empty string, so a single-tenant deployment's keys
/// are as unambiguously namespaced as a multi-tenant one's — and so that a
/// tenant later appearing does not collide with the keys written before it.
///
/// **Never spellable by a real tenant.** A tenant id has no charset
/// restriction, so a literal any tenant could also be named would put a
/// deployment's tenant `_` in the same namespace as its single-tenant
/// requests — reachable in a tenancy migration, or with the plugin router
/// mounted outside the tenancy middleware. Presence is therefore encoded in the
/// segment itself: see [`tenant_segment`].
pub const NO_TENANT: &str = "-";

/// Prefix marking a namespace segment that names a real tenant.
const TENANT_PRESENT: char = 't';

/// The namespace segment for `tenant`, injective in `Option<&str>`.
///
/// A present tenant is `TENANT_PRESENT` followed by its id; absence is
/// [`NO_TENANT`], which no present tenant can produce because every one of them
/// starts with that prefix. This is the whole of "an absent tenant is not a
/// tenant named `_`", and it is a derivation rather than a check for the same
/// reason the physical table name is: the colliding spelling stops existing
/// instead of being refused.
#[must_use]
pub fn tenant_segment(tenant: Option<&str>) -> String {
    tenant.map_or_else(
        || NO_TENANT.to_owned(),
        |id| format!("{TENANT_PRESENT}{id}"),
    )
}

/// Enforcement, once, for every capability.
///
/// Holds the manifest's grants and quotas, the tenant, the backends and the
/// ledger, so that "is this allowed" is answered in one place rather than in
/// five backends that each have to remember.
pub struct CapabilityRuntime {
    /// Read by each capability module to name itself in a derived key or a
    /// job record; never written after construction.
    pub(super) plugin: String,
    /// The sealed digest of the artifact this runtime serves, stamped on every
    /// job it enqueues.
    pub(super) artifact_sha256: String,
    capabilities: Vec<SandboxCapability>,
    /// Read by the outbound module to hand an implementation the host list it
    /// needs for a per-hop decision; never written after construction.
    pub(super) grants: super::grants::CapabilityGrants,
    /// The backends. `pub(super)` so each capability module reaches its own and
    /// no more; nothing outside this module tree can substitute one.
    pub(super) services: CapabilityServices,
    quotas: QuotaLedger,
    rate: Option<Arc<quota::CapabilityRateLimiter>>,
    events: Vec<Event>,
    /// Calls the ledger could not hold, and when, so a truncated summary says
    /// it is one.
    dropped: Option<DroppedCalls>,
    /// `(operation, outcome)` pairs already warned about this request.
    warned: Vec<(&'static str, Outcome)>,
}

/// The most distinct `(operation, outcome)` pairs one request warns about.
///
/// Both sets are closed — ten operations, a handful of outcomes — so this is a
/// ceiling the vocabulary already implies rather than a policy. It exists so the
/// scan above stays bounded if either set grows.
const MAX_WARNED: usize = 64;

/// Truncate a guest-chosen target to [`MAX_TARGET_CHARS`] *characters*.
///
/// By characters, before escaping: `rejected` expands a hostile character to as
/// many as ten bytes, so truncating afterwards would still have materialised the
/// expansion first.
fn bounded_target(target: &str) -> String {
    let mut chars = target.chars();
    let kept: String = chars.by_ref().take(MAX_TARGET_CHARS).collect();
    if chars.next().is_some() {
        format!("{kept}…")
    } else {
        kept
    }
}

/// Characters of a guest-chosen target the audit ledger keeps.
///
/// Deliberately far below the 512 characters the manifest module's `rejected`
/// escaper allows. A ledger entry answers "which table / host / key", and no honest one
/// needs a paragraph — while `rejected` escapes a hostile character to as many
/// as ten bytes, so 512 characters is up to 5 KiB per entry and
/// [`MAX_EVENTS`] of those is megabytes the request footprint never budgeted.
pub const MAX_TARGET_CHARS: usize = 96;

/// The most events one request's ledger holds.
///
/// Bounded for the same reason the denial ledger is: a guest that calls in a
/// loop must not be able to turn the evidence of it into the host's
/// memory-exhaustion channel. The per-request `calls` quota bounds the count
/// long before this does; this is the backstop for a build whose quota was
/// raised.
pub const MAX_EVENTS: usize = 512;

impl fmt::Debug for CapabilityRuntime {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CapabilityRuntime")
            .field("plugin", &self.plugin)
            .field("capabilities", &self.capabilities)
            .field("events", &self.events.len())
            .finish_non_exhaustive()
    }
}

impl CapabilityRuntime {
    /// Build the runtime one request will use.
    #[must_use]
    pub fn new(manifest: &SandboxManifest, services: CapabilityServices) -> Self {
        Self {
            plugin: manifest.name.clone(),
            artifact_sha256: manifest.sha256.clone(),
            capabilities: manifest.capabilities.clone(),
            grants: manifest.grants.clone(),
            rate: services.rate.clone(),
            services,
            quotas: QuotaLedger::new(manifest.quotas),
            events: Vec::new(),
            dropped: None,
            warned: Vec::new(),
        }
    }

    /// Stamp jobs with the sealed digest of the whole artifact.
    ///
    /// Defaults to the module digest, which is all a bare host has.
    #[must_use]
    pub fn with_artifact_sha256(mut self, digest: impl Into<String>) -> Self {
        self.artifact_sha256 = digest.into();
        self
    }

    /// The quotas this runtime enforces.
    #[must_use]
    pub const fn quotas(&self) -> &CapabilityQuotas {
        self.quotas.declared()
    }

    /// The tenant every scoped call binds to, as the caller supplied it.
    ///
    /// `None` when the caller supplied none. That is right for a single-tenant
    /// application and **wrong** for a multi-tenant one whose plugin router was
    /// mounted outside the tenancy middleware: every tenant's keys and rows
    /// would pool into one namespace. Nothing inside the sandbox can tell those
    /// two apart — `CURRENT_TENANT` is simply absent in both — so the caller is
    /// the one that has to know, which is why
    /// [`CapabilityServices::for_tenant`] exists and why
    /// `SandboxedPlugin::serve` reads the task-local on the async task rather
    /// than inside `spawn_blocking`, where it is absent for a third reason.
    #[must_use]
    pub fn tenant(&self) -> Option<&str> {
        self.services.tenant.as_deref()
    }

    /// The namespace segment every scoped derivation uses.
    ///
    /// [`tenant_segment`] of [`tenant`](Self::tenant): the *physical* tenant,
    /// the way [`physical_table`](db::physical_table) is the physical table.
    /// Never the raw id, because the raw id cannot say whether there was one.
    #[must_use]
    pub fn tenant_key(&self) -> String {
        tenant_segment(self.tenant())
    }

    /// Everything this request did, in order.
    #[must_use]
    pub fn events(&self) -> &[Event] {
        &self.events
    }

    /// Take the ledger, leaving it empty.
    #[must_use]
    pub fn take_events(&mut self) -> Vec<Event> {
        std::mem::take(&mut self.events)
    }

    /// Calls this request made that the bounded ledger could not hold.
    ///
    /// Non-zero means the activity summary is a floor, not a count — which is
    /// the one thing an operator must not have to guess at.
    #[must_use]
    pub const fn dropped_events(&self) -> u64 {
        match self.dropped {
            Some(dropped) => dropped.count,
            None => 0,
        }
    }

    /// As [`dropped_events`](Self::dropped_events), with the times of the
    /// first and last dropped call. `None` when no call was dropped.
    ///
    /// Give it to [`PluginActivityLog::ingest_dropped_calls`]. The log then
    /// windows the drops by when they occurred, not by when the request ended.
    #[must_use]
    pub const fn dropped_calls(&self) -> Option<DroppedCalls> {
        self.dropped
    }

    /// The bytes `call`'s reply may take at its worst, under this plugin's
    /// quotas. See [`CapabilityCall::reply_reserve_for`].
    pub(crate) const fn reply_reserve(&self, call: &CapabilityCall) -> usize {
        let quotas = self.quotas();
        let reserve =
            call.reply_reserve_for(quotas.kv_value_bytes as usize, quotas.db_rows as usize);
        match call {
            // The body is at most `outbound_response_bytes`, and one byte
            // escapes to at most six. Smaller than `MAX_RESULT_BYTES` when the
            // plugin asks for small replies.
            CapabilityCall::HttpFetch { .. } => {
                let body = (quotas.outbound_response_bytes as usize).saturating_mul(6);
                reserve.saturating_sub(MAX_RESULT_BYTES.saturating_sub(body))
            }
            _ => reserve,
        }
    }

    /// Whether this plugin may use `capability` at all.
    ///
    /// `is_granted`, not `grants`, for the reason
    /// [`SandboxManifest::is_granted`](super::manifest::SandboxManifest::is_granted)
    /// carries the same name: `grants` is the field holding the scope lists.
    #[must_use]
    pub fn is_granted(&self, capability: SandboxCapability) -> bool {
        self.capabilities.contains(&capability)
    }

    /// Answer one call.
    ///
    /// Never fails and never panics: everything a guest can get wrong comes back
    /// as a [`CallResult::Denied`] it can read.
    pub fn dispatch(&mut self, call: &CapabilityCall) -> CallResult {
        self.dispatch_within(call, usize::MAX)
    }

    /// As [`dispatch`](Self::dispatch), for a caller that has `reply_room`
    /// bytes left to carry the answer.
    ///
    /// A call whose worst-case reply does not fit is refused before it runs.
    /// See [`CapabilityCall::reply_reserve`].
    pub fn dispatch_within(&mut self, call: &CapabilityCall, reply_room: usize) -> CallResult {
        self.dispatch_checked(call, reply_room, |_| true)
    }

    /// As [`dispatch_within`](Self::dispatch_within). `fuel_covers` gets the
    /// worst-case reply in bytes. When it returns `false`, the call is refused
    /// before it runs.
    pub(crate) fn dispatch_checked(
        &mut self,
        call: &CapabilityCall,
        reply_room: usize,
        fuel_covers: impl FnOnce(usize) -> bool,
    ) -> CallResult {
        let id = call.id();
        let capability = call.capability();
        // What the call *names*, computed before any check, so every refusal
        // records what was reached for rather than a dash. The operator surface
        // is built from these, and "this plugin tried to read `users`" is the
        // half of the answer a validated-target-only ledger would drop.
        //
        // Bounded on the way in: `record` truncates and neutralises it, because
        // a KV key is a guest-chosen string that ends up in a log line and in
        // the rendered summary.
        let named = call.target_excerpt();

        // The shared `calls` budget is charged first, for every dispatch —
        // including one that is about to be refused. A refusal is cheaper than
        // a call but it is not free: it costs a grant scan, an encoded reply, a
        // ledger entry and a log line, and leaving refusals unmetered let a
        // plugin with a generous fuel budget emit ~10^5 warn lines per request
        // against a `calls` quota of 128. Fuel alone is the wrong ceiling for
        // work whose cost is a log write.
        //
        // The *per-capability* counters stay where they are, below, so a call
        // refused before it could reach a backend does not spend the budget
        // that bounds backend work.
        if let Err(field) = self.quotas.charge_call() {
            return self.record(
                call,
                &named,
                CallResult::denied(
                    id,
                    DenialReason::QuotaExceeded,
                    format!("the per-request `{field}` quota is spent"),
                ),
            );
        }

        // Then the capability, before the scope: a plugin that was never
        // granted `db` must learn *that*, not that some table is not in a list
        // it does not have.
        if !self.is_granted(capability) {
            return self.record(
                call,
                &named,
                CallResult::denied(
                    id,
                    DenialReason::CapabilityNotGranted,
                    format!("this plugin was not granted `{capability}`"),
                ),
            );
        }

        // Scope before the per-capability charge: an out-of-scope target never
        // reaches a backend, so spending the budget that bounds backend work on
        // it would let one manifest mistake starve the calls the plugin is
        // entitled to make.
        let target = match self.target_of(call) {
            Ok(target) => target,
            Err(result) => return self.record(call, &named, result),
        };

        // Room for the answer, before anything is charged or committed: once a
        // write has happened its reply can no longer be refused.
        let reserve = self.reply_reserve(call);
        if reply_room < reserve {
            return self.record(
                call,
                &target,
                CallResult::denied(
                    id,
                    DenialReason::QuotaExceeded,
                    "there is not room to deliver this call's answer; read the pending replies \
                     and retry",
                ),
            );
        }
        if !fuel_covers(reserve) {
            return self.record(
                call,
                &target,
                CallResult::denied(
                    id,
                    DenialReason::QuotaExceeded,
                    "the fuel left cannot read this call's answer and finish the request; \
                     increase `fuel`",
                ),
            );
        }

        // Shape before the per-capability charge, for the same reason: a call
        // refused for its own malformed content never reaches a backend, and
        // charging `db_writes` for it would deny a valid call that follows.
        // The shared `calls` budget was already charged above.
        if let Err(result) = self.validate(call) {
            return self.record(call, &target, result);
        }

        if let Err(field) = self.quotas.charge_capability(call) {
            return self.record(
                call,
                &target,
                CallResult::denied(
                    id,
                    DenialReason::QuotaExceeded,
                    format!("the per-request `{field}` quota is spent"),
                ),
            );
        }

        // The rate ceiling last of the three, and after the per-request charge:
        // it is the only one that depends on what *other* requests did, so a
        // plugin reading its denials sees "I asked too much" before "everyone
        // asked too much at once", which is the order the two are fixable in.
        if self
            .rate
            .as_ref()
            .is_some_and(|rate| !rate.try_take(capability))
        {
            return self.record(
                call,
                &target,
                CallResult::denied(
                    id,
                    DenialReason::QuotaExceeded,
                    format!("this plugin is over its `{capability}` calls-per-second ceiling"),
                ),
            );
        }

        let result = self.perform(call, &target);
        self.record(call, &target, result)
    }

    /// The thing this call names, checked against the grant list.
    ///
    /// Returns the *logical* target — the same string
    /// [`CapabilityCall::target`] names, once it has been checked against the
    /// grant list. A physical name is derived later and is an implementation
    /// detail nobody auditing the plugin asked about.
    fn target_of(&self, call: &CapabilityCall) -> Result<String, CallResult> {
        let id = call.id();
        match call {
            CapabilityCall::KvGet { key, .. }
            | CapabilityCall::KvSet { key, .. }
            | CapabilityCall::KvDelete { key, .. } => {
                // The one identifier a guest chooses freely, so it is the one
                // with a shape rule of its own. Control characters are refused
                // because the key is printed in the audit surface an operator
                // reads.
                if key.is_empty() || key.len() > MAX_KV_KEY_BYTES {
                    return Err(CallResult::denied(
                        id,
                        DenialReason::Malformed,
                        format!("a kv key must be 1..={MAX_KV_KEY_BYTES} bytes"),
                    ));
                }
                if key.chars().any(char::is_control) {
                    return Err(CallResult::denied(
                        id,
                        DenialReason::Malformed,
                        "a kv key may not carry control characters",
                    ));
                }
                Ok(key.clone())
            }
            CapabilityCall::HttpFetch { url, .. } => {
                let host = outbound::host_of(url).ok_or_else(|| {
                    CallResult::denied(
                        id,
                        DenialReason::Malformed,
                        "expected an absolute http/https URL with a bare host and no userinfo",
                    )
                })?;
                self.require_grant(id, SandboxCapability::HttpOutbound, &host, "hosts")?;
                Ok(host)
            }
            CapabilityCall::DbInsert { table, .. }
            | CapabilityCall::DbGet { table, .. }
            | CapabilityCall::DbQuery { table, .. }
            | CapabilityCall::DbUpdate { table, .. }
            | CapabilityCall::DbDelete { table, .. } => {
                self.require_grant(id, SandboxCapability::Db, table, "tables")?;
                Ok(table.clone())
            }
            CapabilityCall::JobEnqueue { job_type, .. } => {
                self.require_grant(id, SandboxCapability::Jobs, job_type, "job_types")?;
                Ok(job_type.clone())
            }
        }
    }

    fn require_grant(
        &self,
        id: u64,
        capability: SandboxCapability,
        entry: &str,
        field: &'static str,
    ) -> Result<(), CallResult> {
        if self.grants.allows(capability, entry) {
            return Ok(());
        }
        Err(CallResult::denied(
            id,
            DenialReason::NotInGrant,
            format!(
                "`[grants].{field}` does not name {entry:?}",
                entry = super::manifest::rejected(entry)
            ),
        ))
    }

    /// The call's shape and size, checked without copying it or charging a quota.
    fn validate(&self, call: &CapabilityCall) -> Result<(), CallResult> {
        match call {
            CapabilityCall::KvSet { .. } => kv::validate(self, call),
            CapabilityCall::KvGet { .. } | CapabilityCall::KvDelete { .. } => Ok(()),
            CapabilityCall::HttpFetch { .. } => outbound::check_request(self, call).map(|_| ()),
            CapabilityCall::DbInsert { .. }
            | CapabilityCall::DbGet { .. }
            | CapabilityCall::DbQuery { .. }
            | CapabilityCall::DbUpdate { .. }
            | CapabilityCall::DbDelete { .. } => db::validate(call),
            CapabilityCall::JobEnqueue { .. } => jobs::validate(call),
        }
    }

    /// Run the call against its backend, everything already checked.
    fn perform(&self, call: &CapabilityCall, target: &str) -> CallResult {
        match call {
            CapabilityCall::KvGet { .. }
            | CapabilityCall::KvSet { .. }
            | CapabilityCall::KvDelete { .. } => kv::perform(self, call, target),
            CapabilityCall::HttpFetch { .. } => outbound::perform(self, call, target),
            CapabilityCall::DbInsert { .. }
            | CapabilityCall::DbGet { .. }
            | CapabilityCall::DbQuery { .. }
            | CapabilityCall::DbUpdate { .. }
            | CapabilityCall::DbDelete { .. } => db::perform(self, call, target),
            CapabilityCall::JobEnqueue { .. } => jobs::perform(self, call, target),
        }
    }

    /// Record one call and hand the result back unchanged.
    ///
    /// Every exit from `dispatch` goes through here, so there is no path that
    /// answers the guest without also telling the operator.
    fn record(&mut self, call: &CapabilityCall, target: &str, result: CallResult) -> CallResult {
        // Bounded and neutralised once, here, rather than at each of the six
        // exits. A KV key is the one identifier a guest chooses freely, a table
        // or job type is bounded only by the wire, and from here both go into a
        // `tracing` line and the rendered operator summary — which a terminal
        // renders. `MAX_TARGET_CHARS` is far below what `rejected` would allow,
        // because a ledger this deep with 512-character entries is megabytes the
        // footprint never budgeted.
        let target = super::manifest::rejected(&bounded_target(target));
        let outcome = match result.denial() {
            None => Outcome::Allowed,
            Some(DenialReason::QuotaExceeded) => Outcome::QuotaExceeded,
            Some(reason) => Outcome::Denied(reason),
        };
        if matches!(outcome, Outcome::Allowed) {
            tracing::debug!(
                plugin = self.plugin,
                capability = call.capability().as_str(),
                operation = call.operation(),
                target = target.as_str(),
                "sandboxed plugin used a granted capability"
            );
        } else if self
            .warned
            .iter()
            .any(|(operation, seen)| *operation == call.operation() && *seen == outcome)
        {
            // Already said, this request. Deduplicated by `(operation,
            // outcome)` exactly as `HostState::deny` deduplicates a refused
            // import, and for the same reason: a guest that calls in a loop
            // must not be able to turn the evidence of it into the operator's
            // log-storage problem. The ledger below still counts every one, so
            // nothing is lost — only repeated.
        } else {
            // Denials are warned, once, at the point they happen — whoever is
            // driving the host should not have to remember to log it.
            tracing::warn!(
                plugin = self.plugin,
                capability = call.capability().as_str(),
                operation = call.operation(),
                target = target.as_str(),
                outcome = %outcome,
                "sandboxed plugin was refused a capability call"
            );
            // Bounded by the vocabulary: at most one entry per (operation,
            // outcome) pair, and both sets are closed.
            if self.warned.len() < MAX_WARNED {
                self.warned.push((call.operation(), outcome));
            }
        }
        let at = crate::time::ambient_instant();
        if self.events.len() < MAX_EVENTS {
            self.events.push(Event {
                capability: call.capability(),
                operation: call.operation(),
                target,
                outcome,
                at,
            });
        } else {
            // Counted rather than dropped silently. The ledger is bounded
            // because a guest chooses how many entries it makes, but an
            // operator reading "12 db-inserts" off a truncated ledger would be
            // reading a number the plugin chose. `MAX_EVENTS` is not
            // unreachable: the quotas that would bound it live in the plugin's
            // own manifest, so the plugin sets them.
            //
            // Keep the time too. The log ingests the request when it ends. A
            // slow request must not make an old drop look recent (#2523).
            self.dropped = Some(self.dropped.map_or(
                DroppedCalls {
                    count: 1,
                    first: at,
                    last: at,
                },
                |dropped| DroppedCalls {
                    count: dropped.count.saturating_add(1),
                    last: at,
                    ..dropped
                },
            ));
        }
        result
    }
}

#[cfg(test)]
mod tests {
    use super::super::grants::MAX_QUOTA;
    use super::*;
    use jobs::MemoryJobSink;
    use kv::MemoryKvStore;
    use outbound::{OutboundResponse, RecordingHttp};

    /// A manifest granting everything, so each test can turn one thing off
    /// rather than assembling a manifest per case.
    fn manifest(capabilities: &[SandboxCapability]) -> SandboxManifest {
        let caps = capabilities
            .iter()
            .map(|capability| format!("\"{}\"", capability.as_str()))
            .collect::<Vec<_>>()
            .join(", ");
        let mut grants = String::from("[grants]\n");
        if capabilities.contains(&SandboxCapability::HttpOutbound) {
            grants.push_str("hosts = [\"api.example.com\"]\n");
        }
        if capabilities.contains(&SandboxCapability::Db) {
            grants.push_str("tables = [\"orders\"]\n");
        }
        if capabilities.contains(&SandboxCapability::Jobs) {
            grants.push_str("job_types = [\"reindex\"]\n");
        }
        if capabilities.contains(&SandboxCapability::Render) {
            grants.push_str("slots = [\"order-summary\"]\n");
        }
        SandboxManifest::parse(&format!(
            r#"
name = "shop"
version = "0.1.0"
wire_version = 1
prefix = "/shop"
capabilities = [{caps}]
sha256 = "{digest}"

[[routes]]
method = "GET"
path = "/shop/panel"

{grants}
"#,
            digest = "c".repeat(64)
        ))
        .expect("the fixture manifest is valid")
    }

    fn everything() -> Vec<SandboxCapability> {
        SandboxCapability::ALL.to_vec()
    }

    struct Fixture {
        runtime: CapabilityRuntime,
        kv: Arc<MemoryKvStore>,
        db: Arc<db::MemoryPluginStore>,
        jobs: Arc<MemoryJobSink>,
        http: Arc<RecordingHttp>,
    }

    fn fixture(capabilities: &[SandboxCapability], tenant: Option<&str>) -> Fixture {
        let kv = MemoryKvStore::new();
        let db = db::MemoryPluginStore::new();
        let jobs = MemoryJobSink::new();
        let http = RecordingHttp::new();
        let mut services = CapabilityServices {
            kv: Some(Arc::clone(&kv) as Arc<dyn KvStore>),
            db: Some(Arc::clone(&db) as Arc<dyn PluginStore>),
            jobs: Some(Arc::clone(&jobs) as Arc<dyn JobSink>),
            http: Some(Arc::clone(&http) as Arc<dyn OutboundHttp>),
            ..CapabilityServices::none()
        };
        if let Some(tenant) = tenant {
            services = services.for_tenant(tenant);
        }
        Fixture {
            runtime: CapabilityRuntime::new(&manifest(capabilities), services),
            kv,
            db,
            jobs,
            http,
        }
    }

    fn row(pairs: &[(&str, &str)]) -> PluginRow {
        pairs
            .iter()
            .map(|(key, value)| ((*key).to_owned(), PluginValue::Text((*value).to_owned())))
            .collect()
    }

    // ── Results an embedder's backend hands back (issue #2523) ───────

    /// A store that answers with whatever it was given, to stand in for an
    /// embedder's implementation that breaks the trait contract.
    struct RogueStore {
        inserted: String,
        rows: Vec<PluginRow>,
    }

    impl db::PluginStore for RogueStore {
        fn insert(&self, _: &db::Scope, _: PluginRow) -> Result<String, db::StoreError> {
            Ok(self.inserted.clone())
        }
        fn get(&self, _: &db::Scope, _: &str) -> Result<Option<PluginRow>, db::StoreError> {
            Ok(self.rows.first().cloned())
        }
        fn query(
            &self,
            _: &db::Scope,
            _: &PluginRow,
            _: usize,
            _: usize,
            _: Option<&str>,
        ) -> Result<db::QueryPage, db::StoreError> {
            Ok(db::QueryPage {
                rows: self.rows.clone(),
                truncated: false,
            })
        }
        fn update(&self, _: &db::Scope, _: &str, _: PluginRow) -> Result<(), db::StoreError> {
            Ok(())
        }
        fn delete(&self, _: &db::Scope, _: &str) -> Result<(), db::StoreError> {
            Ok(())
        }
    }

    struct RogueSink(String);

    impl JobSink for RogueSink {
        fn enqueue(&self, _: jobs::PluginJob) -> Result<String, String> {
            Ok(self.0.clone())
        }
    }

    fn rogue_db(inserted: &str, rows: Vec<PluginRow>) -> CapabilityRuntime {
        let services = CapabilityServices {
            db: Some(Arc::new(RogueStore {
                inserted: inserted.to_owned(),
                rows,
            }) as Arc<dyn PluginStore>),
            ..CapabilityServices::none()
        };
        CapabilityRuntime::new(&manifest(&everything()), services)
    }

    fn db_get() -> CapabilityCall {
        CapabilityCall::DbGet {
            id: 1,
            table: "orders".to_owned(),
            row_id: "r1".to_owned(),
        }
    }

    #[test]
    fn a_stored_row_over_the_row_ceiling_is_not_served() {
        // Between `MAX_ROW_BYTES` and `MAX_RESULT_BYTES`: `bounded_rows` let it
        // through although the host would refuse the same row on insert.
        let mut fat = row(&[("row_id", "r1")]);
        for column in 0..5 {
            fat.insert(
                format!("c{column}"),
                PluginValue::Text("x".repeat(MAX_VALUE_TEXT_BYTES)),
            );
        }
        assert!(row_weight(&fat) > MAX_ROW_BYTES && row_weight(&fat) < MAX_RESULT_BYTES);
        let mut runtime = rogue_db("r1", vec![fat.clone()]);
        assert_eq!(
            runtime.dispatch(&db_get()).denial(),
            Some(DenialReason::ResultRejected)
        );
        let query = CapabilityCall::DbQuery {
            id: 2,
            table: "orders".to_owned(),
            filter: PluginRow::new(),
            limit: 0,
            after: None,
        };
        let rejected = runtime.dispatch(&query);
        assert_eq!(rejected.denial(), Some(DenialReason::ResultRejected));
        // The row id is named, so the guest can `db-delete` the row.
        assert!(format!("{rejected:?}").contains("r1"), "{rejected:?}");
    }

    #[test]
    fn a_stored_row_the_host_would_not_have_accepted_is_not_served() {
        let bad_name = row(&[("row_id", "r1"), ("Bad-Name", "x")]);
        let host_column = row(&[("row_id", "r1"), (db::TENANT_COLUMN, "alpha")]);
        let too_wide: PluginRow = (0..=MAX_ROW_COLUMNS)
            .map(|index| (format!("c{index}"), PluginValue::Int(1)))
            .collect();
        for stored in [bad_name, host_column, too_wide] {
            let mut runtime = rogue_db("r1", vec![stored]);
            assert_eq!(
                runtime.dispatch(&db_get()).denial(),
                Some(DenialReason::ResultRejected)
            );
        }
    }

    #[test]
    fn a_stored_row_without_a_usable_id_is_not_served() {
        let long = "r".repeat(MAX_ROW_ID_BYTES + 1);
        for id in [
            None,
            Some(PluginValue::Int(7)),
            Some(PluginValue::Text(String::new())),
            Some(PluginValue::Text(long)),
        ] {
            let mut stored = row(&[("sku", "A-1")]);
            if let Some(id) = id {
                stored.insert("row_id".to_owned(), id);
            }
            let mut runtime = rogue_db("r1", vec![stored]);
            assert_eq!(
                runtime.dispatch(&db_get()).denial(),
                Some(DenialReason::ResultRejected)
            );
        }
    }

    #[test]
    fn a_legal_stored_row_is_still_served_with_its_row_id() {
        let mut runtime = rogue_db("r1", vec![row(&[("row_id", "r1"), ("sku", "A-1")])]);
        assert!(runtime.dispatch(&db_get()).denial().is_none());
    }

    #[test]
    fn a_store_assigned_id_the_guest_could_not_use_is_a_rejected_result() {
        for bad in [String::new(), "r".repeat(MAX_ROW_ID_BYTES + 1)] {
            let mut runtime = rogue_db(&bad, Vec::new());
            let result = runtime.dispatch(&CapabilityCall::DbInsert {
                id: 1,
                table: "orders".to_owned(),
                row: row(&[("sku", "A-1")]),
            });
            assert_eq!(
                result.denial(),
                Some(DenialReason::ResultRejected),
                "{} byte id",
                bad.len()
            );
        }
    }

    #[test]
    fn a_sink_assigned_job_id_over_the_ceiling_is_refused_but_counted_as_queued() {
        let services = CapabilityServices {
            jobs: Some(Arc::new(RogueSink("j".repeat(MAX_JOB_ID_BYTES + 1))) as Arc<dyn JobSink>),
            ..CapabilityServices::none()
        };
        let mut runtime = CapabilityRuntime::new(&manifest(&everything()), services);
        let result = runtime.dispatch(&CapabilityCall::JobEnqueue {
            id: 1,
            job_type: "reindex".to_owned(),
            payload: PluginRow::new(),
        });
        assert_eq!(result.denial(), Some(DenialReason::ResultRejected));
        let log = PluginActivityLog::new();
        log.ingest("shop", runtime.take_events());
        let summary = log.summary("shop", std::time::Duration::from_secs(60));
        assert_eq!(
            summary.job_types.get("reindex").copied(),
            Some(1),
            "the enqueue committed: {summary:?}"
        );
    }

    #[test]
    fn a_call_refused_for_its_shape_spends_no_backend_quota() {
        // Default `db_writes` is 32 and `calls` is 128: 40 malformed inserts
        // would have spent the first and denied the valid one that follows.
        let mut fixture = fixture(&everything(), Some("alpha"));
        let fat: PluginRow = (0..=MAX_ROW_COLUMNS)
            .map(|index| (format!("c{index}"), PluginValue::Int(1)))
            .collect();
        for _ in 0..40 {
            let result = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
                id: 1,
                table: "orders".to_owned(),
                row: fat.clone(),
            });
            assert_eq!(result.denial(), Some(DenialReason::Malformed));
        }
        let ok = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
            id: 2,
            table: "orders".to_owned(),
            row: row(&[("sku", "A-1")]),
        });
        assert!(ok.denial().is_none(), "{ok:?}");

        for _ in 0..10 {
            let bad = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
                id: 3,
                method: "TRACE".to_owned(),
                url: "https://api.example.com/".to_owned(),
                headers: Vec::new(),
                body: String::new(),
            });
            assert_eq!(bad.denial(), Some(DenialReason::Malformed));
        }
        let ok = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 4,
            method: "GET".to_owned(),
            url: "https://api.example.com/".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        assert_ne!(ok.denial(), Some(DenialReason::QuotaExceeded), "{ok:?}");
    }

    #[test]
    fn a_huge_target_is_cut_before_it_is_copied() {
        let key = "k".repeat(MAX_KV_KEY_BYTES * 1000);
        let excerpt = CapabilityCall::KvGet { id: 1, key }.target_excerpt();
        assert!(
            excerpt.chars().count() <= MAX_TARGET_CHARS + 1,
            "{}",
            excerpt.len()
        );
        let url = format!("https://{}.test/", "h".repeat(1 << 20));
        let excerpt = CapabilityCall::HttpFetch {
            id: 1,
            method: "GET".to_owned(),
            url,
            headers: Vec::new(),
            body: String::new(),
        }
        .target_excerpt();
        assert!(
            excerpt.chars().count() <= MAX_TARGET_CHARS + 1,
            "{}",
            excerpt.len()
        );
    }

    #[test]
    fn a_row_exactly_at_the_ceiling_is_served_and_one_byte_over_is_not() {
        let at_ceiling = |extra: usize| {
            // 3 names ("c0".."c2", "row_id" excluded from the count) + values.
            let mut stored = row(&[("row_id", "r1")]);
            let mut left = MAX_ROW_BYTES + extra;
            for column in 0..4 {
                let name = format!("c{column}");
                left -= name.len();
                let take = left.min(MAX_VALUE_TEXT_BYTES);
                left -= take;
                stored.insert(name, PluginValue::Text("x".repeat(take)));
            }
            assert_eq!(left, 0);
            stored
        };
        assert!(
            rogue_db("r1", vec![at_ceiling(0)])
                .dispatch(&db_get())
                .denial()
                .is_none()
        );
        assert_eq!(
            rogue_db("r1", vec![at_ceiling(1)])
                .dispatch(&db_get())
                .denial(),
            Some(DenialReason::ResultRejected)
        );
    }

    #[test]
    fn a_job_id_at_its_ceiling_is_returned_and_an_empty_one_is_not() {
        for (id, rejected) in [("j".repeat(MAX_JOB_ID_BYTES), false), (String::new(), true)] {
            let services = CapabilityServices {
                jobs: Some(Arc::new(RogueSink(id)) as Arc<dyn JobSink>),
                ..CapabilityServices::none()
            };
            let mut runtime = CapabilityRuntime::new(&manifest(&everything()), services);
            let result = runtime.dispatch(&CapabilityCall::JobEnqueue {
                id: 1,
                job_type: "reindex".to_owned(),
                payload: PluginRow::new(),
            });
            assert_eq!(
                result.denial() == Some(DenialReason::ResultRejected),
                rejected,
                "{result:?}"
            );
        }
    }

    #[test]
    fn a_query_reserve_follows_its_page_and_fits_the_queue() {
        let query = |limit| CapabilityCall::DbQuery {
            id: 1,
            table: "orders".to_owned(),
            filter: PluginRow::new(),
            limit,
            after: None,
        };
        // A huge quota must not make every query undeliverable.
        let huge = 1_000_000;
        assert!(query(1).reply_reserve_for(0, huge) < query(0).reply_reserve_for(0, huge));
        assert!(query(0).reply_reserve_for(0, huge) <= super::super::host::MAX_QUEUED_REPLY_BYTES);
        // A page of one row reserves one row of framing.
        assert!(query(1).reply_reserve_for(0, huge) < MAX_RESULT_BYTES + 8192);
    }

    #[test]
    fn reply_room_is_checked_at_its_exact_edge() {
        let post = CapabilityCall::HttpFetch {
            id: 1,
            method: "POST".to_owned(),
            url: "https://api.example.com/".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        };
        let reserve = post.reply_reserve();
        let mut fixture = fixture(&everything(), Some("alpha"));
        let short = fixture.runtime.dispatch_within(&post, reserve - 1);
        assert_eq!(short.denial(), Some(DenialReason::QuotaExceeded));
        assert!(fixture.http.seen().is_empty(), "the POST ran");
        let fits = fixture.runtime.dispatch_within(&post, reserve);
        assert_ne!(fits.denial(), Some(DenialReason::QuotaExceeded), "{fits:?}");
        // A read reserves room too.
        let get = CapabilityCall::KvGet {
            id: 2,
            key: "cart".to_owned(),
        };
        let read = get.reply_reserve_for(fixture.runtime.quotas().kv_value_bytes as usize, 0);
        assert_eq!(
            fixture.runtime.dispatch_within(&get, read - 1).denial(),
            Some(DenialReason::QuotaExceeded)
        );
        assert!(
            fixture
                .runtime
                .dispatch_within(&get, read)
                .denial()
                .is_none()
        );
    }

    #[test]
    fn an_oversized_kv_value_is_refused_without_spending_the_write_quota() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        let huge = PluginValue::Text("x".repeat(64 * 1024 + 1));
        for _ in 0..40 {
            let result = fixture.runtime.dispatch(&CapabilityCall::KvSet {
                id: 1,
                key: "cart".to_owned(),
                value: huge.clone(),
            });
            assert_eq!(result.denial(), Some(DenialReason::QuotaExceeded));
        }
        // 40 refusals, `kv_writes` is 32: a valid write still succeeds.
        let ok = fixture.runtime.dispatch(&CapabilityCall::KvSet {
            id: 2,
            key: "cart".to_owned(),
            value: PluginValue::Int(1),
        });
        assert!(ok.denial().is_none(), "{ok:?}");
    }

    // ── Capability gating ────────────────────────────────────────────

    #[test]
    fn an_ungranted_capability_is_denied_and_recorded() {
        let mut fixture = fixture(&[SandboxCapability::HttpRequest], None);
        let result = fixture.runtime.dispatch(&CapabilityCall::KvGet {
            id: 1,
            key: "cart".to_owned(),
        });
        assert_eq!(result.denial(), Some(DenialReason::CapabilityNotGranted));
        assert_eq!(fixture.runtime.events().len(), 1);
        assert_eq!(
            fixture.runtime.events().first().map(|event| event.outcome),
            Some(audit::CapabilityOutcome::Denied(
                DenialReason::CapabilityNotGranted
            ))
        );
        assert!(fixture.kv.keys().is_empty(), "nothing was written");
    }

    #[test]
    fn a_refusal_records_what_was_reached_for_not_a_dash() {
        // "What did this plugin do" includes what it *tried* to do. A ledger
        // that named only the targets that passed would answer the easy half.
        let mut granted = fixture(&everything(), Some("alpha"));
        let _ = granted.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "GET".to_owned(),
            url: "https://attacker.test/steal".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        let _ = granted.runtime.dispatch(&CapabilityCall::DbQuery {
            id: 2,
            table: "users".to_owned(),
            filter: PluginRow::new(),
            limit: 1,
            after: None,
        });
        let _ = granted.runtime.dispatch(&CapabilityCall::JobEnqueue {
            id: 3,
            job_type: "drain-accounts".to_owned(),
            payload: PluginRow::new(),
        });
        let targets: Vec<&str> = granted
            .runtime
            .events()
            .iter()
            .map(|event| event.target.as_str())
            .collect();
        assert_eq!(targets, vec!["attacker.test", "users", "drain-accounts"]);

        // And it is neutralised on the way in: the one identifier a guest
        // chooses freely ends up in a log line and in the operator summary.
        let mut hostile = fixture(&[SandboxCapability::HttpRequest], None);
        let _ = hostile.runtime.dispatch(&CapabilityCall::KvGet {
            id: 4,
            key: "a\u{1b}[2K\rforged".to_owned(),
        });
        let recorded = hostile
            .runtime
            .events()
            .first()
            .map(|event| event.target.clone())
            .unwrap_or_default();
        assert!(!recorded.contains('\u{1b}'), "{recorded:?}");
    }

    #[test]
    fn an_ungranted_call_cannot_spend_the_shared_call_budget() {
        // The order of the checks is load-bearing: if a call to an ungranted
        // capability charged the shared `calls` quota first, a plugin could be
        // starved of the surface it *was* granted by one it was not.
        let mut ungranted = fixture(&[SandboxCapability::HttpRequest], None);
        for id in 0..1_000 {
            let _ = ungranted.runtime.dispatch(&CapabilityCall::KvGet {
                id,
                key: "cart".to_owned(),
            });
        }
        // The ledger is bounded, but nothing was ever charged: a granted call
        // on a fresh runtime with the same quotas still succeeds.
        let mut granted = fixture(&everything(), None);
        assert!(
            granted
                .runtime
                .dispatch(&CapabilityCall::KvGet {
                    id: 1,
                    key: "cart".to_owned(),
                })
                .denial()
                .is_none()
        );
    }

    // ── KV containment ───────────────────────────────────────────────

    /// A `KvStore` that is down rather than empty.
    #[derive(Debug)]
    struct UnreachableKvStore;

    impl KvStore for UnreachableKvStore {
        fn get(&self, _key: &str) -> Result<Option<PluginValue>, String> {
            Err("the key/value backend is unreachable".to_owned())
        }

        fn set(&self, _key: &str, _value: PluginValue) -> Result<(), String> {
            Err("the key/value backend is unreachable".to_owned())
        }

        fn delete(&self, _key: &str) -> Result<(), String> {
            Err("the key/value backend is unreachable".to_owned())
        }
    }

    #[test]
    fn a_kv_backend_outage_reaches_the_guest_as_a_failure_not_a_miss() {
        // `found: false` is an answer about the *data*; an outage is an answer
        // about the *backend*. Collapsing the second into the first tells a
        // plugin its cart is empty when the truth is that nobody could look,
        // and the audit ledger then records a read that never happened.
        let manifest = manifest(&everything());
        let mut runtime = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                kv: Some(Arc::new(UnreachableKvStore) as Arc<dyn KvStore>),
                ..CapabilityServices::none()
            }
            .for_tenant("alpha"),
        );

        let result = runtime.dispatch(&CapabilityCall::KvGet {
            id: 1,
            key: "cart".to_owned(),
        });
        assert!(
            matches!(
                result,
                CallResult::Denied {
                    id: 1,
                    reason: DenialReason::BackendError,
                    ..
                }
            ),
            "an unreachable store must deny, not report an absent key: {result:?}"
        );

        // And a delete that did not happen must not read as `Done` either: a
        // plugin releasing a lock would carry on over a value still there.
        let result = runtime.dispatch(&CapabilityCall::KvDelete {
            id: 2,
            key: "cart".to_owned(),
        });
        assert!(
            matches!(
                result,
                CallResult::Denied {
                    id: 2,
                    reason: DenialReason::BackendError,
                    ..
                }
            ),
            "an unreachable store must deny the delete too: {result:?}"
        );
    }

    #[test]
    fn an_overlong_outbound_method_is_refused_without_being_copied() {
        // The method was `to_ascii_uppercase`d before anything decided it was
        // a method, so a frame carrying a megabyte of "method" was copied for
        // the fold and again for the denial — ahead of the check that refuses
        // it, and outside `outbound_request_bytes`, which is measured later
        // and does not count the method. The match is now allocation-free and
        // length-first, so this is refused in constant time.
        // A client has to be wired, or the call is refused as `Unavailable`
        // before the method is looked at — which would have made this test pass
        // for the wrong reason. It caught exactly that on its first run.
        let manifest = manifest(&everything());
        let http = RecordingHttp::new();
        let mut runtime = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                http: Some(Arc::clone(&http) as Arc<dyn OutboundHttp>),
                ..CapabilityServices::none()
            }
            .for_tenant("alpha"),
        );

        let result = runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "G".repeat(1024 * 1024),
            url: "https://api.example.com/orders".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        assert!(
            matches!(
                result,
                CallResult::Denied {
                    id: 1,
                    reason: DenialReason::Malformed,
                    ..
                }
            ),
            "an overlong method is not a method: {result:?}"
        );

        // And the match stays case-insensitive, which is what the fold was for.
        let lowercase = runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 2,
            method: "post".to_owned(),
            url: "https://api.example.com/orders".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        assert!(
            !matches!(
                lowercase,
                CallResult::Denied {
                    reason: DenialReason::Malformed,
                    ..
                }
            ),
            "`post` must still be POST: {lowercase:?}"
        );
    }

    #[test]
    fn one_tenants_key_is_unreadable_when_another_is_active() {
        let store = MemoryKvStore::new();
        let services = |tenant: &str| {
            CapabilityServices {
                kv: Some(Arc::clone(&store) as Arc<dyn KvStore>),
                ..CapabilityServices::none()
            }
            .for_tenant(tenant)
        };

        let manifest = manifest(&everything());
        let mut alpha = CapabilityRuntime::new(&manifest, services("alpha"));
        let _ = alpha.dispatch(&CapabilityCall::KvSet {
            id: 1,
            key: "cart".to_owned(),
            value: PluginValue::Text("alpha's".to_owned()),
        });

        let mut beta = CapabilityRuntime::new(&manifest, services("beta"));
        let result = beta.dispatch(&CapabilityCall::KvGet {
            id: 2,
            key: "cart".to_owned(),
        });
        assert_eq!(
            result,
            CallResult::Ok {
                id: 2,
                value: CallValue::Value {
                    value: PluginValue::Null,
                    found: false
                }
            },
            "tenant beta must not see tenant alpha's key"
        );
    }

    #[test]
    fn no_tenant_key_can_be_spelled_to_reach_another_tenants_namespace() {
        // The escape a naive `format!("{plugin}:{tenant}:{key}")` would allow:
        // tenant `a` asking for the key `b:cart` lands where tenant `b` writes
        // `cart`. Both segments are escaped, so the two keys stay distinct.
        let store = MemoryKvStore::new();
        let manifest = manifest(&everything());
        let services = |tenant: &str| {
            CapabilityServices {
                kv: Some(Arc::clone(&store) as Arc<dyn KvStore>),
                ..CapabilityServices::none()
            }
            .for_tenant(tenant)
        };

        let mut beta = CapabilityRuntime::new(&manifest, services("b"));
        let _ = beta.dispatch(&CapabilityCall::KvSet {
            id: 1,
            key: "cart".to_owned(),
            value: PluginValue::Text("beta's".to_owned()),
        });

        let mut alpha = CapabilityRuntime::new(&manifest, services("a"));
        let result = alpha.dispatch(&CapabilityCall::KvGet {
            id: 2,
            key: "b:cart".to_owned(),
        });
        assert_eq!(
            result,
            CallResult::Ok {
                id: 2,
                value: CallValue::Value {
                    value: PluginValue::Null,
                    found: false
                }
            }
        );
        assert_eq!(store.keys().len(), 1, "{:?}", store.keys());
    }

    #[test]
    fn a_tenant_id_carrying_the_separator_cannot_collide_with_another() {
        // Tenant ids are the application's, not the manifest's: they come from
        // a header or a subdomain, and nothing validates their charset.
        assert_ne!(
            kv::namespaced_key("shop", Some("b%3Ax"), "k"),
            kv::namespaced_key("shop", Some("b"), "%3Ax:k")
        );
        assert_ne!(
            kv::namespaced_key("shop", Some("a"), "b:c"),
            kv::namespaced_key("shop", Some("a:b"), "c")
        );
    }

    #[test]
    fn two_plugins_never_share_a_key() {
        assert_ne!(
            kv::namespaced_key("shop", Some("t"), "cart"),
            kv::namespaced_key("other", Some("t"), "cart")
        );
    }

    #[test]
    fn no_tenant_is_not_a_tenant_that_could_be_named() {
        // A tenant id has no charset restriction, so the single-tenant sentinel
        // must be a segment no tenant could also be. It used to be `_`, which a
        // deployment can name a tenant — and then that tenant shared a
        // namespace with every request that arrived without one, reachable in a
        // tenancy migration or with the router mounted outside the tenancy
        // middleware.
        for spelling in [NO_TENANT, "_", "", "-", "t", "t-"] {
            assert_ne!(
                tenant_segment(Some(spelling)),
                tenant_segment(None),
                "a tenant named {spelling:?} reached the single-tenant namespace"
            );
            assert_ne!(
                kv::namespaced_key("shop", Some(spelling), "cart"),
                kv::namespaced_key("shop", None, "cart"),
                "{spelling:?}"
            );
        }
        // And distinct tenants stay distinct, which is the property the marker
        // must not have broken.
        assert_ne!(tenant_segment(Some("a")), tenant_segment(Some("b")));
    }

    #[test]
    fn a_stored_value_over_the_current_ceiling_is_refused_on_the_way_out() {
        // Lowering `kv_value_bytes` is meant to reduce authority, but the store
        // keeps what the old ceiling allowed. Serving it anyway would make the
        // tightened quota cosmetic, and a large enough value would overrun the
        // reply queue and fail the request rather than being refused.
        let kv = MemoryKvStore::new();
        let manifest = manifest(&everything());
        let key = kv::namespaced_key(&manifest.name, Some("alpha"), "cart");
        let _ = kv.set(&key, PluginValue::Text("x".repeat(4096)));

        let mut quotas = manifest.quotas;
        quotas.kv_value_bytes = 64;
        let tightened = SandboxManifest { quotas, ..manifest };
        let mut runtime = CapabilityRuntime::new(
            &tightened,
            CapabilityServices {
                kv: Some(Arc::clone(&kv) as Arc<dyn KvStore>),
                ..CapabilityServices::none()
            }
            .for_tenant("alpha"),
        );
        let result = runtime.dispatch(&CapabilityCall::KvGet {
            id: 1,
            key: "cart".to_owned(),
        });
        assert_eq!(
            result.denial(),
            Some(DenialReason::QuotaExceeded),
            "{result:?}"
        );
    }

    #[test]
    fn a_query_filter_may_not_carry_the_row_id_that_would_widen_it() {
        // `validated_row` strips `row_id` on a write, which is right: a row read
        // back carries its id and echoing it means nothing. On a *filter* the
        // same strip turns "the row with this id" into "every row this tenant
        // has" — narrowing that silently widens, which is the one direction a
        // containment bug can go and still look like it worked.
        let mut fixture = fixture(&everything(), Some("alpha"));
        for sku in ["A-1", "A-2", "A-3"] {
            let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
                id: 1,
                table: "orders".to_owned(),
                row: row(&[("sku", sku)]),
            });
            assert!(inserted.denial().is_none(), "{inserted:?}");
        }
        let result = fixture.runtime.dispatch(&CapabilityCall::DbQuery {
            id: 2,
            table: "orders".to_owned(),
            filter: row(&[(db::ID_COLUMN, "r1")]),
            limit: 0,
            after: None,
        });
        assert_eq!(result.denial(), Some(DenialReason::Malformed), "{result:?}");
    }

    #[test]
    fn a_truncated_page_can_be_continued_from_where_it_stopped() {
        // A ceiling the guest is told about and cannot act on is worse than no
        // ceiling: `truncated` would say "there is more" while the same filter
        // returned the same prefix forever.
        let mut fixture = fixture(&everything(), Some("alpha"));
        for sku in ["A-1", "A-2", "A-3", "A-4"] {
            let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
                id: 1,
                table: "orders".to_owned(),
                row: row(&[("sku", sku)]),
            });
            assert!(inserted.denial().is_none(), "{inserted:?}");
        }

        let page = |runtime: &mut CapabilityRuntime, after: Option<String>| {
            let result = runtime.dispatch(&CapabilityCall::DbQuery {
                id: 2,
                table: "orders".to_owned(),
                filter: PluginRow::new(),
                limit: 2,
                after,
            });
            let CallResult::Ok {
                value: CallValue::Rows { rows, .. },
                ..
            } = result
            else {
                panic!("the query should have succeeded: {result:?}")
            };
            rows
        };

        let first = page(&mut fixture.runtime, None);
        assert_eq!(first.len(), 2, "{first:?}");
        let last = first
            .last()
            .and_then(|row| row.get(db::ID_COLUMN))
            .map(ToString::to_string)
            .expect("every returned row carries its id");
        let second = page(&mut fixture.runtime, Some(last));
        assert_eq!(second.len(), 2, "{second:?}");
        // Disjoint, which is the whole property: a cursor that returned the
        // same rows again would page forever without advancing.
        for row in &second {
            assert!(!first.contains(row), "{row:?} came back twice");
        }
    }

    #[test]
    fn a_row_at_the_column_ceiling_survives_a_read_modify_write() {
        // The documented pattern: read a row, change a field, send it back. The
        // row comes back carrying the `row_id` the host added, so a row stored
        // at exactly `MAX_ROW_COLUMNS` arrived at `db-update` with one column
        // too many — refused for a column the guest never wrote. The ceilings
        // must be a function of what the plugin stored, not of where the row
        // had been.
        let mut fixture = fixture(&everything(), Some("alpha"));
        let full: PluginRow = (0..MAX_ROW_COLUMNS)
            .map(|index| {
                (
                    format!("c{index}"),
                    PluginValue::Int(i64::try_from(index).unwrap_or(0)),
                )
            })
            .collect();
        assert_eq!(full.len(), MAX_ROW_COLUMNS);

        let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
            id: 1,
            table: "orders".to_owned(),
            row: full,
        });
        let CallResult::Ok {
            value: CallValue::RowId { row_id },
            ..
        } = inserted
        else {
            panic!("a row at the ceiling should store: {inserted:?}")
        };

        let read = fixture.runtime.dispatch(&CapabilityCall::DbGet {
            id: 2,
            table: "orders".to_owned(),
            row_id: row_id.clone(),
        });
        let CallResult::Ok {
            value: CallValue::Rows { rows, .. },
            ..
        } = read
        else {
            panic!("the read should have succeeded: {read:?}")
        };
        let mut echoed = rows.first().cloned().unwrap_or_default();
        assert_eq!(
            echoed.len(),
            MAX_ROW_COLUMNS + 1,
            "the row comes back with the host's id, which is the whole problem"
        );
        echoed.insert("c0".to_owned(), PluginValue::Int(99));

        let updated = fixture.runtime.dispatch(&CapabilityCall::DbUpdate {
            id: 3,
            table: "orders".to_owned(),
            row_id,
            row: echoed,
        });
        assert!(
            updated.denial().is_none(),
            "writing back what was read must not be refused: {updated:?}"
        );
    }

    #[test]
    fn a_page_cut_by_the_row_quota_says_so_too() {
        // `truncated` is the guest's only signal that more rows exist. A page
        // cut by `db_rows` looked complete, and with `limit: 0` the guest is
        // never told the quota — so a full page and a finished table were
        // indistinguishable, and the cursor documented for continuing was never
        // reached.
        let base = manifest(&everything());
        let mut quotas = base.quotas;
        quotas.db_rows = 2;
        let manifest = SandboxManifest { quotas, ..base };
        let store = db::MemoryPluginStore::new();
        let mut runtime = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                db: Some(Arc::clone(&store) as Arc<dyn PluginStore>),
                ..CapabilityServices::none()
            }
            .for_tenant("alpha"),
        );
        for sku in ["A-1", "A-2", "A-3", "A-4"] {
            let inserted = runtime.dispatch(&CapabilityCall::DbInsert {
                id: 1,
                table: "orders".to_owned(),
                row: row(&[("sku", sku)]),
            });
            assert!(inserted.denial().is_none(), "{inserted:?}");
        }

        let result = runtime.dispatch(&CapabilityCall::DbQuery {
            id: 2,
            table: "orders".to_owned(),
            filter: PluginRow::new(),
            limit: 0,
            after: None,
        });
        let CallResult::Ok {
            value: CallValue::Rows { rows, truncated },
            ..
        } = result
        else {
            panic!("the query should have succeeded: {result:?}")
        };
        assert_eq!(rows.len(), 2, "the quota caps the page");
        assert!(truncated, "and the guest is told there is more");
    }

    #[test]
    fn a_read_modify_write_echo_still_parses_at_the_column_ceiling() {
        // The round-trip this asserts was already fixed once, at the dispatch
        // layer, and the parse-time column bound reintroduced it one layer
        // earlier — where the existing test could not see it, because that one
        // builds a `CapabilityCall` directly instead of parsing a frame.
        //
        // `db-get` returns the row plus the host's `row_id`, so a row stored at
        // exactly `MAX_ROW_COLUMNS` comes back with one more entry than a row
        // may have, and sending it to `db-update` is the documented workflow.
        let columns = (0..MAX_ROW_COLUMNS)
            .map(|index| format!(r#""c{index}":"v""#))
            .chain(std::iter::once(r#""row_id":"r1""#.to_owned()))
            .collect::<Vec<_>>()
            .join(",");
        let line = format!(
            r#"{{"call":"db-update","id":1,"table":"orders","row_id":"r1","row":{{{columns}}}}}"#
        );
        let parsed = serde_json::from_str::<CapabilityCall>(&line);
        assert!(
            parsed.is_ok(),
            "the echo of a full row must survive parsing: {parsed:?}"
        );

        // And one genuine column past the ceiling is still refused while read.
        let columns = (0..=MAX_ROW_COLUMNS)
            .map(|index| format!(r#""c{index}":"v""#))
            .chain(std::iter::once(r#""row_id":"r1""#.to_owned()))
            .collect::<Vec<_>>()
            .join(",");
        let line = format!(
            r#"{{"call":"db-update","id":1,"table":"orders","row_id":"r1","row":{{{columns}}}}}"#
        );
        assert!(
            serde_json::from_str::<CapabilityCall>(&line).is_err(),
            "a row genuinely over the ceiling is still refused"
        );
    }

    #[test]
    fn the_root_of_a_fragment_is_bounded_like_its_branches() {
        // The branches were bounded first and the root was left open, which
        // made bounding the branches pointless: `render` applies `MAX_NODES` to
        // a tree serde has already built, and the host then clones it.
        let nodes = (0..=render::MAX_NODES)
            .map(|_| r#"{"node":"text","text":""}"#)
            .collect::<Vec<_>>()
            .join(",");
        let line = format!(r#"{{"op":"fragment","nodes":[{nodes}]}}"#);
        let err = serde_json::from_str::<crate::plugin_sandbox::wire::GuestFrame>(&line)
            .expect_err("a fragment over the node ceiling must not parse");
        assert!(err.to_string().contains("more than"), "{err}");

        // Exactly at the ceiling still parses.
        let nodes = (0..render::MAX_NODES)
            .map(|_| r#"{"node":"text","text":""}"#)
            .collect::<Vec<_>>()
            .join(",");
        let line = format!(r#"{{"op":"fragment","nodes":[{nodes}]}}"#);
        assert!(
            serde_json::from_str::<crate::plugin_sandbox::wire::GuestFrame>(&line).is_ok(),
            "a fragment at the ceiling is legal"
        );
    }

    #[test]
    fn a_guest_collection_is_refused_while_it_is_read_not_after() {
        // The distinction this asserts is invisible from the outside if you
        // only check that an oversized frame is refused: the *old* code also
        // refused it, having first allocated the whole thing. What is being
        // tested is that the refusal comes from the parser, so nothing larger
        // than the ceiling is ever held.
        let pairs = (0..=MAX_OUTBOUND_HEADERS)
            .map(|_| r#"["",""]"#)
            .collect::<Vec<_>>()
            .join(",");
        let line = format!(
            r#"{{"call":"http-fetch","id":1,"method":"GET","url":"https://api.example.com/","headers":[{pairs}],"body":""}}"#
        );
        let parsed = serde_json::from_str::<CapabilityCall>(&line);
        let err = parsed.expect_err("a frame over the header ceiling must not parse");
        assert!(
            err.to_string().contains("more than"),
            "the parser should say why: {err}"
        );

        // And exactly at the ceiling still parses, so the bound is not off by
        // one in the direction that breaks a legal call.
        let pairs = (0..MAX_OUTBOUND_HEADERS)
            .map(|_| r#"["a","b"]"#)
            .collect::<Vec<_>>()
            .join(",");
        let line = format!(
            r#"{{"call":"http-fetch","id":1,"method":"GET","url":"https://api.example.com/","headers":[{pairs}],"body":""}}"#
        );
        assert!(
            serde_json::from_str::<CapabilityCall>(&line).is_ok(),
            "a frame at the ceiling is legal"
        );
    }

    // ── Outbound containment ─────────────────────────────────────────

    #[test]
    fn a_declared_host_is_reachable_through_the_framework_client() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        fixture.http.answer(
            "https://api.example.com/v1/orders",
            OutboundResponse {
                status: 200,
                headers: vec![("content-type".to_owned(), "application/json".to_owned())],
                body: "[]".to_owned(),
                final_url: "https://api.example.com/v1/orders".to_owned(),
            },
        );
        let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "get".to_owned(),
            url: "https://api.example.com/v1/orders".to_owned(),
            headers: vec![("accept".to_owned(), "application/json".to_owned())],
            body: String::new(),
        });
        assert_eq!(
            result,
            CallResult::Ok {
                id: 1,
                value: CallValue::Http {
                    status: 200,
                    headers: vec![("content-type".to_owned(), "application/json".to_owned())],
                    body: "[]".to_owned(),
                }
            }
        );
        assert_eq!(fixture.http.seen().len(), 1);
    }

    #[test]
    fn every_shape_of_undeclared_host_is_denied_and_never_leaves() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        for url in [
            "https://attacker.test/",
            "https://api.example.com.attacker.test/",
            "https://evil-api.example.com/",
            "https://api.example.com@attacker.test/",
            "https://attacker.test/?to=api.example.com",
            "http://attacker.test/api.example.com",
            "file:///etc/passwd",
            "//api.example.com/",
            "/relative",
            "https://API.EXAMPLE.COM.attacker.test/",
            "https://[::1]/",
        ] {
            let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
                id: 9,
                method: "GET".to_owned(),
                url: url.to_owned(),
                headers: Vec::new(),
                body: String::new(),
            });
            assert!(
                matches!(
                    result.denial(),
                    Some(DenialReason::NotInGrant | DenialReason::Malformed)
                ),
                "{url} was allowed: {result:?}"
            );
        }
        assert!(
            fixture.http.seen().is_empty(),
            "not one of those may have left: {:?}",
            fixture.http.seen()
        );
    }

    #[test]
    fn a_redirect_off_the_granted_host_is_refused_after_the_fact() {
        // Checking the URL the guest wrote bounds the first hop and nothing
        // else. A client that follows redirects — which most do by default —
        // would otherwise send the body to a host the manifest never named and
        // record one allowed call to the granted one.
        let mut fixture = fixture(&everything(), Some("alpha"));
        fixture
            .http
            .answer_from("https://api.example.com/r", "https://attacker.test/collect");
        let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "POST".to_owned(),
            url: "https://api.example.com/r".to_owned(),
            headers: Vec::new(),
            body: "tenant data".to_owned(),
        });
        assert_eq!(
            result.denial(),
            Some(DenialReason::NotInGrant),
            "{result:?}"
        );

        // …and the operator sees where it ended up, not just that something
        // was refused.
        let events = fixture.runtime.events();
        assert!(
            events
                .iter()
                .any(|event| event.outcome != audit::CapabilityOutcome::Allowed),
            "{events:?}"
        );
    }

    #[test]
    fn an_implementation_is_handed_what_it_needs_to_refuse_a_hop_itself() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        fixture.http.answer(
            "https://api.example.com/v1",
            OutboundResponse::from_url("https://api.example.com/v1", 200, "ok"),
        );
        let _ = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "GET".to_owned(),
            url: "https://api.example.com/v1".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        let seen = fixture.http.seen();
        let sent = seen.first().expect("the call left");
        assert!(
            !sent.follow_redirects,
            "redirects are never followed for a plugin"
        );
        assert_eq!(sent.allowed_hosts, vec!["api.example.com".to_owned()]);
        assert!(
            sent.timeout > std::time::Duration::ZERO,
            "a call has a deadline"
        );
    }

    #[test]
    fn an_upstream_that_answered_counts_as_a_host_that_was_called() {
        // A response discarded for being over a byte ceiling still *left the
        // host*. Filing it under "refused" would make the audit surface
        // undercount exactly the calls an operator most wants to see.
        let mut manifest = manifest(&everything());
        manifest.quotas.outbound_response_bytes = 4;
        let http = RecordingHttp::new();
        http.answer(
            "https://api.example.com/big",
            OutboundResponse::from_url("https://api.example.com/big", 200, "far too long"),
        );
        let mut runtime = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                http: Some(Arc::clone(&http) as Arc<dyn OutboundHttp>),
                ..CapabilityServices::none()
            },
        );
        let result = runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "GET".to_owned(),
            url: "https://api.example.com/big".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        assert_eq!(result.denial(), Some(DenialReason::ResponseTooLarge));

        let log = audit::PluginActivityLog::new();
        log.ingest("shop", runtime.take_events());
        let summary = log.summary("shop", std::time::Duration::from_secs(3600));
        assert_eq!(summary.hosts.get("api.example.com"), Some(&1));
        assert!(summary.refused_targets.is_empty(), "{summary:?}");
    }

    #[test]
    fn the_same_host_on_another_port_is_the_granted_host() {
        // The grant is about where the bytes go, and the name decides that.
        assert_eq!(
            outbound::host_of("https://api.example.com:8443/v1"),
            Some("api.example.com".to_owned())
        );
        assert_eq!(
            outbound::host_of("https://api.example.com:not-a-port/"),
            None
        );
    }

    #[test]
    fn a_header_a_plugin_may_not_set_is_denied_before_the_call_leaves() {
        // A fresh fixture per case, so each refusal is read on its own.
        for (name, value) in [
            ("authorization", "Bearer stolen"),
            ("cookie", "session=stolen"),
            ("host", "attacker.test"),
            ("x-forwarded-for", "10.0.0.1"),
        ] {
            let mut one = fixture(&everything(), Some("alpha"));
            let result = one.runtime.dispatch(&CapabilityCall::HttpFetch {
                id: 1,
                method: "GET".to_owned(),
                url: "https://api.example.com/".to_owned(),
                headers: vec![(name.to_owned(), value.to_owned())],
                body: String::new(),
            });
            assert_eq!(result.denial(), Some(DenialReason::NotInGrant), "{name}");
            assert!(one.http.seen().is_empty(), "{name} left the host");
        }

        let mut fixture = fixture(&everything(), Some("alpha"));
        let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 2,
            method: "GET".to_owned(),
            url: "https://api.example.com/".to_owned(),
            headers: vec![("accept".to_owned(), "a\r\nx-evil: 1".to_owned())],
            body: String::new(),
        });
        assert_eq!(result.denial(), Some(DenialReason::Malformed));
        assert!(fixture.http.seen().is_empty());
    }

    #[test]
    fn a_row_no_reply_could_carry_is_refused_where_it_goes_in() {
        // `MAX_VALUE_TEXT_BYTES` bounds one column and 32 of them bounds a row
        // at 2 MiB — twice what one reply may be. Accepting such a row would
        // store something `db-get` could never hand back, so the ceiling that
        // makes a read answerable belongs on the write.
        let mut fixture = fixture(&everything(), Some("alpha"));
        let fat: PluginRow = (0..8)
            .map(|index| {
                (
                    format!("c{index}"),
                    PluginValue::Text("x".repeat(MAX_VALUE_TEXT_BYTES)),
                )
            })
            .collect();
        assert!(
            row_weight(&fat) > MAX_ROW_BYTES,
            "the fixture is fat enough"
        );
        let result = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
            id: 1,
            table: "orders".to_owned(),
            row: fat,
        });
        assert_eq!(result.denial(), Some(DenialReason::Malformed), "{result:?}");
        assert!(fixture.db.keys().is_empty(), "{:?}", fixture.db.keys());
    }

    #[test]
    fn a_query_is_bounded_by_bytes_and_says_when_it_was_cut() {
        // The finding: `db_rows` defaults to 500 and a row may be
        // `MAX_ROW_BYTES`, so a granted plugin could make the host materialise,
        // encode and hold a hundred megabytes per request against a quota its
        // own manifest sets. The budget now reaches the store, and the guest is
        // told its answer was cut rather than reading a short page as the end
        // of the table.
        let mut fixture = fixture(&everything(), Some("alpha"));
        let wide = "x".repeat(MAX_VALUE_TEXT_BYTES);
        for index in 0..16 {
            let mut row = row(&[("kind", "wide")]);
            // Three, not four: four maxed-out columns plus the names is over
            // `MAX_ROW_BYTES`, and a row that cannot be stored cannot
            // demonstrate an answer that is too big to return.
            for column in 0..3 {
                row.insert(format!("c{column}"), PluginValue::Text(wide.clone()));
            }
            row.insert("n".to_owned(), PluginValue::Int(index));
            let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
                id: 1,
                table: "orders".to_owned(),
                row,
            });
            assert!(inserted.denial().is_none(), "row {index}: {inserted:?}");
        }

        let result = fixture.runtime.dispatch(&CapabilityCall::DbQuery {
            id: 2,
            table: "orders".to_owned(),
            filter: row(&[("kind", "wide")]),
            limit: 0,
            after: None,
        });
        let CallResult::Ok {
            value: CallValue::Rows { rows, truncated },
            ..
        } = result
        else {
            panic!("the query should have succeeded: {result:?}")
        };
        assert!(truncated, "16 wide rows do not fit one reply");
        assert!(!rows.is_empty(), "and the answer is not empty either");
        let carried: usize = rows.iter().map(row_weight).sum();
        assert!(
            carried <= MAX_RESULT_BYTES.saturating_add(MAX_ROW_BYTES),
            "the answer carried {carried} bytes"
        );
    }

    #[test]
    fn an_upstream_cannot_answer_with_unbounded_headers() {
        // The allow-list says which response headers come back and nothing
        // about their size, and every one of them is the *upstream's* choice
        // rather than the plugin's or the host's. A legal body plus megabytes
        // of `etag` is an oversized allocation the host makes and then fails
        // the plugin's request over.
        let mut fixture = fixture(&everything(), Some("alpha"));
        let mut answer = OutboundResponse::from_url("https://api.example.com/", 200, "ok");
        answer.headers = (0..64)
            .map(|_| ("etag".to_owned(), "x".repeat(4096)))
            .collect();
        fixture.http.answer("https://api.example.com/", answer);

        let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "GET".to_owned(),
            url: "https://api.example.com/".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        let CallResult::Ok {
            value: CallValue::Http { headers, .. },
            ..
        } = result
        else {
            panic!("the fetch should have succeeded: {result:?}")
        };
        assert!(headers.len() <= MAX_OUTBOUND_HEADERS, "{}", headers.len());
        let carried: usize = headers
            .iter()
            .map(|(name, value)| name.len().saturating_add(value.len()))
            .sum();
        assert!(
            carried <= MAX_RESPONSE_HEADER_BYTES,
            "the reply carried {carried} bytes of headers"
        );
    }

    #[test]
    fn an_outbound_request_body_is_bounded_like_the_response_is() {
        // The quota table bounded what an upstream may *return* and said nothing
        // about what the plugin may *send*. A body is guest-chosen, so its only
        // bound was the stdout frame it arrived in — megabytes by default. An
        // operator granting a hostname agreed to calls, not to unbounded egress
        // through them.
        let mut fixture = fixture(&everything(), Some("alpha"));
        let ceiling = fixture.runtime.quotas().outbound_request_bytes as usize;
        fixture.http.answer(
            "https://api.example.com/",
            OutboundResponse::from_url("https://api.example.com/", 200, "ok"),
        );

        let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "POST".to_owned(),
            url: "https://api.example.com/".to_owned(),
            headers: Vec::new(),
            body: "x".repeat(ceiling + 1),
        });
        assert_eq!(
            result.denial(),
            Some(DenialReason::QuotaExceeded),
            "{result:?}"
        );
        // Refused before dispatch: nothing reached the backend.
        assert!(
            fixture.http.seen().is_empty(),
            "the call still left the host"
        );
    }

    #[test]
    fn an_outbound_requests_headers_count_against_its_byte_quota() {
        // Bounding the body alone left the quota sidesteppable: the allow-list
        // caps how *many* headers may be set and says nothing about their size,
        // so one `user-agent` could carry what a body was just refused for.
        let mut fixture = fixture(&everything(), Some("alpha"));
        let ceiling = fixture.runtime.quotas().outbound_request_bytes as usize;
        fixture.http.answer(
            "https://api.example.com/",
            OutboundResponse::from_url("https://api.example.com/", 200, "ok"),
        );

        let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "POST".to_owned(),
            url: "https://api.example.com/".to_owned(),
            headers: vec![("user-agent".to_owned(), "x".repeat(ceiling + 1))],
            body: String::new(),
        });
        assert_eq!(
            result.denial(),
            Some(DenialReason::QuotaExceeded),
            "an empty body does not make an oversized request small: {result:?}"
        );
        assert!(
            fixture.http.seen().is_empty(),
            "the call still left the host"
        );
    }

    #[test]
    fn the_in_memory_kv_store_is_bounded_by_bytes_not_only_by_keys() {
        // A key count is not a memory bound: the default capacity at the
        // default `kv_value_bytes` is hundreds of megabytes, and gigabytes at
        // the largest legal quota.
        let store = MemoryKvStore::with_capacities(10_000, 4096);
        // Well inside the key ceiling, well past the byte one.
        let mut refused = false;
        for index in 0..64 {
            if store
                .set(&format!("k{index}"), PluginValue::Text("x".repeat(512)))
                .is_err()
            {
                refused = true;
                break;
            }
        }
        assert!(refused, "the store grew past its byte ceiling");
        assert!(
            store.keys().len() < 10_000,
            "and it refused long before the key ceiling"
        );

        // A replacement is checked too: no new key, but more bytes.
        let small = MemoryKvStore::with_capacities(10_000, 4096);
        assert!(small.set("k", PluginValue::Text("x".repeat(64))).is_ok());
        assert!(
            small.set("k", PluginValue::Text("x".repeat(8192))).is_err(),
            "overwriting one key with an oversized value adds no entry and still grows the store"
        );
    }

    #[test]
    fn every_in_memory_backend_is_bounded_by_bytes_not_only_by_count() {
        // The same defect in three places, so it is asserted in one test: a
        // count ceiling times the largest legal entry is gigabytes for the row
        // store, hundreds of megabytes for the job queue. Whatever the count
        // allows, the byte ceiling has to stop first when entries are large.
        let scope = db::Scope {
            table: "plugin_shop__orders".to_owned(),
            tenant: tenant_segment(Some("alpha")),
        };
        let store = db::MemoryPluginStore::with_capacities(10_000, 8192);
        let fat = row(&[("blob", "x".repeat(2048).as_str())]);
        let mut refused = false;
        for _ in 0..64 {
            if store.insert(&scope, fat.clone()).is_err() {
                refused = true;
                break;
            }
        }
        assert!(refused, "the row store grew past its byte ceiling");
        assert!(
            store.keys().len() < 10_000,
            "and refused far below its row ceiling"
        );

        // Replacement is the path that adds no entry and still grows the store,
        // so every backend that has one is asserted here rather than left to be
        // found a round later: `db-update` was bounded only on insert, while
        // `kv-set` already accounted for it.
        let roomy = db::MemoryPluginStore::with_capacities(10_000, 8192);
        let small = roomy
            .insert(&scope, row(&[("blob", "x")]))
            .expect("a small row fits");
        assert!(
            roomy
                .update(&scope, &small, row(&[("blob", "x".repeat(65536).as_str())]))
                .is_err(),
            "replacing a small row with an oversized one adds no row and still grows the store"
        );
        let kv = MemoryKvStore::with_capacities(10_000, 4096);
        assert!(kv.set("k", PluginValue::Text("x".repeat(64))).is_ok());
        assert!(
            kv.set("k", PluginValue::Text("x".repeat(8192))).is_err(),
            "the same must hold for the key/value store"
        );

        let sink = MemoryJobSink::with_capacities(1024, 8192);
        let mut refused = false;
        for _ in 0..64 {
            let job = PluginJob {
                plugin: "shop".to_owned(),
                artifact_sha256: "c".repeat(64),
                tenant: Some("alpha".to_owned()),
                job_type: "reindex".to_owned(),
                payload: row(&[("blob", "x".repeat(2048).as_str())]),
            };
            if sink.enqueue(job).is_err() {
                refused = true;
                break;
            }
        }
        assert!(refused, "the job queue grew past its byte ceiling");
        assert!(
            sink.queued().len() < 1024,
            "and refused far below its depth ceiling"
        );
    }

    // ── DB containment ───────────────────────────────────────────────

    #[test]
    fn a_plugin_reads_and_writes_only_its_own_tenant_scoped_table() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
            id: 1,
            table: "orders".to_owned(),
            row: row(&[("sku", "A-1")]),
        });
        let CallResult::Ok {
            value: CallValue::RowId { row_id },
            ..
        } = inserted
        else {
            panic!("the insert should have succeeded: {inserted:?}")
        };

        // The physical row landed under the derived name, never under a name
        // the guest chose.
        // Derived, never spelled: a literal here is a test that silently stops
        // testing anything the moment the derivation changes.
        let orders = db::physical_table("shop", "orders").expect("derivable");
        assert_eq!(
            fixture.db.keys(),
            vec![(orders, tenant_segment(Some("alpha")), row_id.clone())]
        );

        let read = fixture.runtime.dispatch(&CapabilityCall::DbGet {
            id: 2,
            table: "orders".to_owned(),
            row_id,
        });
        let CallResult::Ok {
            value: CallValue::Rows { rows, .. },
            ..
        } = read
        else {
            panic!("the read should have succeeded: {read:?}")
        };
        assert_eq!(rows.len(), 1);
    }

    #[test]
    fn an_undeclared_table_is_denied_however_it_is_spelled() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        for table in [
            "users",
            "orders_secret",
            "public.users",
            "orders; drop table users",
            "\"users\"",
            "Orders",
            "plugin_shop_orders",
            "",
        ] {
            let result = fixture.runtime.dispatch(&CapabilityCall::DbQuery {
                id: 1,
                table: table.to_owned(),
                filter: PluginRow::new(),
                limit: 10,
                after: None,
            });
            assert_eq!(
                result.denial(),
                Some(DenialReason::NotInGrant),
                "{table} was allowed"
            );
        }
    }

    #[test]
    fn the_node_ceiling_counts_the_whole_tree_not_each_vector() {
        // Third pass over this ground, and the one that makes the count mean
        // what it says. Bounding each `children` vector at `MAX_NODES`
        // separately still admitted many such vectors, so a frame well inside
        // the stdout budget deserialized far more than `MAX_NODES` nodes and
        // was refused only afterwards, on the aggregate, by `render` — after
        // `serde_json` had built the whole tree.
        //
        // Wide and shallow on purpose. Deep nesting is refused by
        // `serde_json`'s own recursion limit long before any of this is
        // reached, so a deep tree would pass whether or not the budget is
        // shared — which is exactly what the first version of this test did.
        let leaves = std::iter::repeat_n(r#"{"node":"text","text":"x"}"#, 8)
            .collect::<Vec<_>>()
            .join(",");
        let branch = format!(r#"{{"node":"element","tag":"p","children":[{leaves}]}}"#);
        // 128 branches of 9 nodes each: every vector is inside `MAX_NODES`, and
        // the tree is over four times it.
        let wide = std::iter::repeat_n(branch.as_str(), 128)
            .collect::<Vec<_>>()
            .join(",");
        let line = format!(r#"{{"op":"fragment","nodes":[{wide}]}}"#);
        let parsed = crate::plugin_sandbox::wire::from_line::<
            crate::plugin_sandbox::wire::GuestFrame,
        >(&line);
        assert!(
            parsed.is_err(),
            "a tree past the node ceiling must be refused while it is read, not after"
        );

        // A tree inside the ceiling still parses, so the budget is not simply
        // refusing anything with children.
        let small = std::iter::repeat_n(branch.as_str(), 4)
            .collect::<Vec<_>>()
            .join(",");
        let line = format!(r#"{{"op":"fragment","nodes":[{small}]}}"#);
        assert!(
            crate::plugin_sandbox::wire::from_line::<crate::plugin_sandbox::wire::GuestFrame>(
                &line
            )
            .is_ok(),
            "a fragment inside the ceiling must still parse"
        );

        // Twice on the same thread: the budget is thread-local, so the failed
        // parse above must not have left this one poorer.
        assert!(
            crate::plugin_sandbox::wire::from_line::<crate::plugin_sandbox::wire::GuestFrame>(
                &line
            )
            .is_ok(),
            "the budget must be restored after each parse"
        );
    }

    #[test]
    fn a_page_of_rows_encodes_within_the_result_ceiling() {
        // `row_weight` counts names and values, not the quotes, colons and
        // commas around them. Many narrow rows then encoded well past
        // `MAX_RESULT_BYTES`, and a reply the reserve had promised to fit was
        // refused after the read.
        let row: PluginRow = (0..MAX_ROW_COLUMNS)
            .map(|i| (format!("c{i:02}"), PluginValue::Text("v".to_owned())))
            .collect();
        let rows = vec![row; 20_000];
        let (kept, truncated) = bounded_rows(rows);
        assert!(truncated);
        let encoded = serde_json::to_vec(&kept).expect("encodes").len();
        assert!(
            encoded <= MAX_RESULT_BYTES,
            "{} rows encode to {encoded} bytes",
            kept.len()
        );

        // Text that grows when escaped: a control character takes six bytes.
        let row: PluginRow = (0..MAX_ROW_COLUMNS)
            .map(|i| (format!("c{i:02}"), PluginValue::Text("\u{1}\"".to_owned())))
            .collect();
        let (kept, _) = bounded_rows(vec![row; 20_000]);
        let encoded = serde_json::to_vec(&kept).expect("encodes").len();
        assert!(
            encoded <= MAX_RESULT_BYTES,
            "escaped rows encode to {encoded}"
        );
    }

    #[test]
    fn a_scalar_weighs_what_it_encodes_to() {
        // A flat eight bytes for every non-text scalar was wrong in the one
        // direction a ceiling may not be wrong in: `i64::MIN` encodes as
        // twenty, so a `kv_value_bytes` of eight admitted a value the operator
        // forbade — on the write, and again on the read, which is checked
        // against the same number.
        assert_eq!(PluginValue::Int(i64::MIN).weight(), 20);
        assert_eq!(PluginValue::Int(i64::MAX).weight(), 19);
        assert_eq!(PluginValue::Int(0).weight(), 1);
        assert_eq!(PluginValue::Int(-1).weight(), 2);
        assert_eq!(PluginValue::Int(999).weight(), 3);
        assert_eq!(PluginValue::Int(-1000).weight(), 5);
        assert_eq!(PluginValue::Null.weight(), 4);
        assert_eq!(PluginValue::Bool(true).weight(), 4);
        assert_eq!(PluginValue::Bool(false).weight(), 5);

        // Against the encoder rather than against arithmetic, so the two cannot
        // drift: no scalar may weigh less than it serializes to.
        for value in [
            PluginValue::Null,
            PluginValue::Bool(true),
            PluginValue::Bool(false),
            PluginValue::Int(0),
            PluginValue::Int(-1),
            PluginValue::Int(i64::MIN),
            PluginValue::Int(i64::MAX),
            PluginValue::Float(-2.225_073_858_507_201_4e-308),
            PluginValue::Float(1.0),
        ] {
            let encoded = serde_json::to_string(&value).expect("encodes");
            assert!(
                value.weight() >= encoded.len(),
                "{value:?} weighs {} but encodes to {} bytes",
                value.weight(),
                encoded.len()
            );
        }
    }

    #[test]
    fn an_oversized_first_row_is_refused_by_the_store_too() {
        // `bounded_rows` dropped its "keep the first one" exemption a while
        // back; the store's own query loop kept its copy of the same
        // exemption, so an oversized row was cloned in full under the store's
        // mutex on every query that matched it and only then discarded a layer
        // up. The clone was the cost, and a guest could repeat it at will.
        let store = db::MemoryPluginStore::new();
        let orders = db::physical_table("shop", "orders").expect("derivable");
        let tenant = tenant_segment(Some("alpha"));
        // Seeded past what `check_row` would accept, which is the only way such
        // a row exists at all.
        let huge = "x".repeat(super::MAX_RESULT_BYTES + 1);
        store.seed(&orders, &tenant, "r1", row(&[("blob", huge.as_str())]));

        let page = store
            .query(
                &db::Scope {
                    table: orders,
                    tenant,
                },
                &PluginRow::new(),
                10,
                super::MAX_RESULT_BYTES,
                None,
            )
            .expect("the store answers");
        assert!(
            page.rows.is_empty(),
            "a row over the whole result budget must not be cloned out of the store"
        );
        assert!(
            page.truncated,
            "and the caller must be told there was something it did not get"
        );
    }

    #[test]
    fn a_generated_row_id_never_lands_on_a_seeded_one() {
        // `seed` places a row at an id of the caller's choosing and cannot
        // advance the insert counter, because that id need not be a number. So
        // a store seeded with `r1` handed `r1` back out on its first insert:
        // `HashMap::insert` replaced the seeded row while both weights had been
        // charged, losing a row with no error and leaving bytes charged against
        // capacity that nothing occupies.
        let store = db::MemoryPluginStore::new();
        let orders = db::physical_table("shop", "orders").expect("derivable");
        let tenant = tenant_segment(Some("alpha"));
        store.seed(&orders, &tenant, "r1", row(&[("sku", "seeded")]));

        let manifest = manifest(&everything());
        let mut runtime = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                db: Some(Arc::clone(&store) as Arc<dyn PluginStore>),
                ..CapabilityServices::none()
            }
            .for_tenant("alpha"),
        );

        let inserted = runtime.dispatch(&CapabilityCall::DbInsert {
            id: 1,
            table: "orders".to_owned(),
            row: row(&[("sku", "inserted")]),
        });
        assert!(
            matches!(inserted, CallResult::Ok { .. }),
            "the insert must succeed: {inserted:?}"
        );

        // Both rows must still be there. The seeded one disappearing is the
        // defect; counting them is what proves it did not.
        let keys = store.keys();
        assert_eq!(
            keys.len(),
            2,
            "the seeded row must survive an insert beside it: {keys:?}"
        );
        assert!(
            keys.iter().any(|(_, _, id)| id == "r1"),
            "the seeded id must still be occupied by its own row: {keys:?}"
        );
    }

    #[test]
    fn another_tenants_rows_are_invisible_and_unwritable() {
        let store = db::MemoryPluginStore::new();
        // Seeded directly: a host-application table, and another tenant's row
        // in the plugin's own table.
        store.seed(
            "users",
            &tenant_segment(Some("alpha")),
            "u1",
            row(&[("email", "ceo@example.com")]),
        );
        // Derived, not hard-coded: a literal here turns the whole test vacuous
        // the moment the derivation changes, because beta's row would land in a
        // table alpha was never going to reach anyway.
        let orders = db::physical_table("shop", "orders").expect("derivable");
        store.seed(
            &orders,
            &tenant_segment(Some("beta")),
            "r99",
            row(&[("sku", "beta's")]),
        );

        let manifest = manifest(&everything());
        let mut alpha = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                db: Some(Arc::clone(&store) as Arc<dyn PluginStore>),
                ..CapabilityServices::none()
            }
            .for_tenant("alpha"),
        );

        let listed = alpha.dispatch(&CapabilityCall::DbQuery {
            id: 1,
            table: "orders".to_owned(),
            filter: PluginRow::new(),
            limit: 100,
            after: None,
        });
        assert_eq!(
            listed,
            CallResult::Ok {
                id: 1,
                value: CallValue::Rows {
                    rows: Vec::new(),
                    truncated: false
                }
            },
            "an unfiltered query returns this tenant's rows, which are none"
        );

        for call in [
            CapabilityCall::DbGet {
                id: 2,
                table: "orders".to_owned(),
                row_id: "r99".to_owned(),
            },
            CapabilityCall::DbUpdate {
                id: 3,
                table: "orders".to_owned(),
                row_id: "r99".to_owned(),
                row: row(&[("sku", "stolen")]),
            },
            CapabilityCall::DbDelete {
                id: 4,
                table: "orders".to_owned(),
                row_id: "r99".to_owned(),
            },
        ] {
            let result = alpha.dispatch(&call);
            // Each call gets the *specific* answer it should, rather than
            // "anything except a populated row": a loop that accepts a denial
            // or an empty read for every operation passes even if `db-delete`
            // started returning `Done` after deleting beta's row.
            let expected_empty = matches!(call, CapabilityCall::DbGet { .. });
            if expected_empty {
                assert_eq!(
                    result,
                    CallResult::Ok {
                        id: result.id(),
                        value: CallValue::Rows {
                            rows: Vec::new(),
                            truncated: false
                        }
                    },
                    "a read of another tenant's id must miss, not error"
                );
            } else {
                assert_eq!(
                    result.denial(),
                    Some(DenialReason::BackendError),
                    "{call:?} must not touch another tenant's row"
                );
            }
        }
        // Untouched, in every sense.
        assert_eq!(
            store.keys(),
            vec![
                (orders, tenant_segment(Some("beta")), "r99".to_owned()),
                (
                    "users".to_owned(),
                    tenant_segment(Some("alpha")),
                    "u1".to_owned()
                ),
            ]
        );
    }

    #[test]
    fn a_row_may_not_carry_the_column_that_decides_its_tenant() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        let result = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
            id: 1,
            table: "orders".to_owned(),
            row: row(&[("tenant_id", "beta")]),
        });
        assert_eq!(result.denial(), Some(DenialReason::NotInGrant));
    }

    #[test]
    fn a_row_id_echoed_back_is_stripped_rather_than_refused() {
        // `db-get` stamps `row_id` onto the row it hands back, so the obvious
        // read-modify-write echoes it. Refusing that would make the obvious
        // code the wrong code; the id is the row's address and travels in its
        // own field, so a value for it in the row conveys nothing and can
        // override nothing. `tenant_id` stays a hard refusal above, because a
        // row that could set *that* would choose its own tenant.
        let mut fixture = fixture(&everything(), Some("alpha"));
        let inserted = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
            id: 1,
            table: "orders".to_owned(),
            row: row(&[("sku", "A-1"), ("row_id", "r999")]),
        });
        let CallResult::Ok {
            value: CallValue::RowId { row_id },
            ..
        } = inserted
        else {
            panic!("the insert should have succeeded: {inserted:?}")
        };
        assert_ne!(row_id, "r999", "the store assigns the id, never the guest");

        let read = fixture.runtime.dispatch(&CapabilityCall::DbGet {
            id: 2,
            table: "orders".to_owned(),
            row_id: row_id.clone(),
        });
        let CallResult::Ok {
            value: CallValue::Rows { rows, .. },
            ..
        } = read
        else {
            panic!("the read should have succeeded: {read:?}")
        };
        let stored = rows.first().cloned().unwrap_or_default();
        assert_eq!(
            stored.get("row_id"),
            Some(&PluginValue::Text(row_id.clone())),
            "the row comes back addressed by the id the store assigned"
        );

        // …and that same row writes straight back.
        let updated = fixture.runtime.dispatch(&CapabilityCall::DbUpdate {
            id: 3,
            table: "orders".to_owned(),
            row_id,
            row: stored,
        });
        assert!(updated.denial().is_none(), "{updated:?}");
    }

    #[test]
    fn two_plugins_can_never_be_named_onto_one_physical_table() {
        // The collision that matters is not punctuation, it is the *separator*:
        // a hostile author picks a plugin name that shifts the boundary onto a
        // victim's table. Both manifests validate, both consent screens are
        // truthful, and `AppBuilder` sees two distinct plugin names.
        let victim = db::physical_table("shop", "orders_v2");
        let attacker = db::physical_table("shop_orders", "v2");
        assert!(victim.is_some() && attacker.is_some());
        assert_ne!(victim, attacker, "a boundary shift must not collide");

        // Punctuation and case are the same bug one variation along.
        let names = ["my-shop", "my.shop", "my_shop", "My_Shop", "MY.SHOP"];
        let derived: Vec<String> = names
            .iter()
            .filter_map(|name| db::physical_table(name, "orders"))
            .collect();
        assert_eq!(derived.len(), names.len(), "{derived:?}");
        let mut unique = derived.clone();
        unique.sort();
        unique.dedup();
        assert_eq!(unique.len(), names.len(), "collision in {derived:?}");

        // And the derived name is still a bare, unquoted SQL identifier.
        for name in &derived {
            assert!(
                name.starts_with("plugin_")
                    && name.bytes().all(|byte| byte.is_ascii_lowercase()
                        || byte.is_ascii_digit()
                        || byte == b'_'),
                "{name}"
            );
        }
    }

    #[test]
    fn a_truncated_ledger_says_so_rather_than_reporting_a_smaller_number() {
        // `MAX_EVENTS` is reachable: the quotas that would bound it live in the
        // plugin's own manifest, so the plugin sets them. An operator reading
        // "12 db-inserts" off a truncated ledger would be reading a number the
        // plugin chose.
        let mut manifest = manifest(&everything());
        manifest.quotas.calls = MAX_QUOTA;
        manifest.quotas.kv_reads = MAX_QUOTA;
        let mut runtime = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
                ..CapabilityServices::none()
            },
        );
        for id in 0..(MAX_EVENTS as u64 + 25) {
            let _ = runtime.dispatch(&CapabilityCall::KvGet {
                id,
                key: "k".to_owned(),
            });
        }
        assert_eq!(runtime.events().len(), MAX_EVENTS);
        assert_eq!(runtime.dropped_events(), 25);

        let log = audit::PluginActivityLog::new();
        let dropped = runtime.dropped_events();
        log.ingest("shop", runtime.take_events());
        log.ingest_dropped("shop", dropped);
        let summary = log.summary("shop", std::time::Duration::from_secs(3600));
        assert_eq!(summary.dropped, 25);
        let rendered = summary.to_string();
        assert!(rendered.contains("floor"), "{rendered}");
    }

    #[test]
    fn a_dropped_call_keeps_its_own_time_not_the_time_of_ingest() {
        // #2523 finding 18. The log ingests a request when it ends. A slow
        // request must not make its old drops look recent.
        use chrono::TimeZone as _;

        use crate::time::{TickingClock, install_ambient};

        let hour = std::time::Duration::from_secs(3600);
        let clock = TickingClock::starting_at(
            chrono::Utc
                .with_ymd_and_hms(2020, 1, 1, 0, 0, 0)
                .single()
                .expect("a valid date"),
        );
        let _clock = install_ambient(Arc::new(clock.clone()));
        let mut manifest = manifest(&everything());
        manifest.quotas.calls = MAX_QUOTA;
        manifest.quotas.kv_reads = MAX_QUOTA;
        let mut runtime = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
                ..CapabilityServices::none()
            },
        );
        let log = audit::PluginActivityLog::new();
        let mut kv_get = (0..).map(|id| CapabilityCall::KvGet {
            id,
            key: "k".to_owned(),
        });
        let start = crate::time::ambient_instant();
        for call in kv_get.by_ref().take(MAX_EVENTS + 20) {
            let _ = runtime.dispatch(&call);
        }
        // Five more drops 30 minutes later.
        clock.advance(hour / 2);
        let later = crate::time::ambient_instant();
        for call in kv_get.take(5) {
            let _ = runtime.dispatch(&call);
        }
        // The request ends 90 minutes after that.
        clock.advance(3 * hour / 2);
        let dropped = runtime.dropped_calls().expect("25 calls were dropped");
        assert_eq!(dropped.count, 25);
        assert_eq!(dropped.first, start);
        assert_eq!(dropped.last, later);
        log.ingest("shop", runtime.take_events());
        log.ingest_dropped_calls("shop", dropped);
        assert_eq!(log.summary("shop", hour).dropped, 0);
        // The window holds the last drop, not the first.
        assert_eq!(log.summary("shop", 7 * hour / 4).dropped, 25);
    }

    #[test]
    fn the_outbound_reply_reserve_follows_the_response_quota() {
        let post = CapabilityCall::HttpFetch {
            id: 1,
            method: "POST".to_owned(),
            url: "https://api.example.com/".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        };
        let fixture = fixture(&everything(), Some("alpha"));
        // The default quota escapes past `MAX_RESULT_BYTES`, so that bound holds.
        assert_eq!(fixture.runtime.reply_reserve(&post), post.reply_reserve());
        let mut manifest = manifest(&everything());
        manifest.quotas.outbound_response_bytes = 4096;
        let small = CapabilityRuntime::new(&manifest, CapabilityServices::none());
        assert_eq!(
            small.reply_reserve(&post),
            6 * 4096 + outbound::MAX_RESPONSE_HEADER_BYTES * 6 + 4096
        );
    }

    #[test]
    fn a_call_commits_when_it_can_change_state_outside_the_guest() {
        let fetch = |method: &str| CapabilityCall::HttpFetch {
            id: 1,
            method: method.to_owned(),
            url: "https://api.example.com/".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        };
        let table = || "orders".to_owned();
        let commits = [
            CapabilityCall::KvSet {
                id: 1,
                key: "k".to_owned(),
                value: PluginValue::Null,
            },
            CapabilityCall::KvDelete {
                id: 1,
                key: "k".to_owned(),
            },
            CapabilityCall::DbInsert {
                id: 1,
                table: table(),
                row: PluginRow::new(),
            },
            CapabilityCall::DbUpdate {
                id: 1,
                table: table(),
                row_id: "r1".to_owned(),
                row: PluginRow::new(),
            },
            CapabilityCall::DbDelete {
                id: 1,
                table: table(),
                row_id: "r1".to_owned(),
            },
            CapabilityCall::JobEnqueue {
                id: 1,
                job_type: "reindex".to_owned(),
                payload: PluginRow::new(),
            },
            fetch("POST"),
            fetch("PUT"),
            fetch("PATCH"),
            fetch("DELETE"),
            // An unknown method is refused later. Here it counts as a write.
            fetch("BREW"),
        ];
        for call in &commits {
            assert!(call.commits(), "{call:?}");
        }
        let reads = [
            CapabilityCall::KvGet {
                id: 1,
                key: "k".to_owned(),
            },
            CapabilityCall::DbGet {
                id: 1,
                table: table(),
                row_id: "r1".to_owned(),
            },
            CapabilityCall::DbQuery {
                id: 1,
                table: table(),
                filter: PluginRow::new(),
                limit: 0,
                after: None,
            },
            fetch("GET"),
            fetch("HEAD"),
            fetch("get"),
        ];
        for call in &reads {
            assert!(!call.commits(), "{call:?}");
        }
    }

    #[test]
    fn an_audit_target_is_bounded_before_it_is_escaped() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        let _ = fixture.runtime.dispatch(&CapabilityCall::KvGet {
            id: 1,
            key: "\u{202e}".repeat(400),
        });
        let recorded = fixture
            .runtime
            .events()
            .first()
            .map(|event| event.target.clone())
            .unwrap_or_default();
        // `rejected` expands a hostile character to as many as ten bytes, so
        // the bound has to be on characters and has to be applied first.
        assert!(
            recorded.len() <= MAX_TARGET_CHARS * 10 + 8,
            "{} bytes: {recorded}",
            recorded.len()
        );
    }

    #[test]
    fn the_in_memory_stores_refuse_rather_than_growing_without_a_ceiling() {
        // A plugin declares its own `kv_writes` and `db_writes`, and both may
        // legally be `MAX_QUOTA`. A shipped store with no ceiling of its own
        // turns that into the host's memory.
        let kv = MemoryKvStore::with_capacity(2);
        assert!(kv.set("a", PluginValue::Int(1)).is_ok());
        assert!(kv.set("b", PluginValue::Int(1)).is_ok());
        assert!(kv.set("c", PluginValue::Int(1)).is_err());
        // Overwriting an existing key adds nothing, so it is never refused.
        assert!(kv.set("a", PluginValue::Int(2)).is_ok());

        let store = db::MemoryPluginStore::with_capacity(1);
        let scope = db::Scope {
            table: "plugin_shop__orders".to_owned(),
            tenant: "alpha".to_owned(),
        };
        assert!(store.insert(&scope, PluginRow::new()).is_ok());
        assert!(store.insert(&scope, PluginRow::new()).is_err());
    }

    #[test]
    fn a_full_store_is_a_denial_the_guest_can_read() {
        let mut runtime = CapabilityRuntime::new(
            &manifest(&everything()),
            CapabilityServices {
                kv: Some(MemoryKvStore::with_capacity(0) as Arc<dyn KvStore>),
                ..CapabilityServices::none()
            }
            .for_tenant("alpha"),
        );
        assert_eq!(
            runtime
                .dispatch(&CapabilityCall::KvSet {
                    id: 1,
                    key: "cart".to_owned(),
                    value: PluginValue::Int(1)
                })
                .denial(),
            Some(DenialReason::BackendError)
        );
    }

    #[test]
    fn no_physical_table_is_derivable_from_a_name_the_grant_list_could_not_hold() {
        assert_eq!(
            db::physical_table("shop", "orders"),
            Some("plugin_shop__orders".to_owned())
        );
        // A plugin name's charset is wider than SQL's, so it is escaped
        // injectively rather than folded: `.` is `_2e` and `-` is `_2d`, and
        // `__` — which no escape can produce — is the separator.
        assert_eq!(
            db::physical_table("my.shop-v2", "orders"),
            Some("plugin_my_2eshop_2dv2__orders".to_owned())
        );
        for (plugin, table) in [
            ("shop", "users; drop table x"),
            ("shop", "\"users\""),
            ("shop", "Orders"),
            ("shop\"; --", "orders"),
            ("shop", ""),
            ("", "orders"),
            ("shop", &"o".repeat(64)),
            // Long, but each half is legal: the *derived* name is what cannot
            // fit, and `SandboxManifest::validate` refuses this at load so an
            // operator never approves a capability that can only be denied.
            (&"p".repeat(40), &"t".repeat(30)),
        ] {
            assert_eq!(db::physical_table(plugin, table), None, "{plugin}/{table}");
        }
    }

    // ── Jobs ─────────────────────────────────────────────────────────

    #[test]
    fn a_job_carries_the_enqueuing_plugin_and_tenant_not_the_guests_word_for_them() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        let result = fixture.runtime.dispatch(&CapabilityCall::JobEnqueue {
            id: 1,
            job_type: "reindex".to_owned(),
            payload: row(&[("since", "2026-01-01")]),
        });
        assert!(result.denial().is_none(), "{result:?}");
        assert_eq!(
            fixture.jobs.queued(),
            vec![jobs::PluginJob {
                plugin: "shop".to_owned(),
                artifact_sha256: "c".repeat(64),
                job_type: "reindex".to_owned(),
                tenant: Some("alpha".to_owned()),
                payload: row(&[("since", "2026-01-01")]),
            }]
        );
    }

    #[test]
    fn an_undeclared_job_type_never_reaches_the_queue() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        for job_type in ["send-invoice", "reindex-all", "REINDEX", ""] {
            let result = fixture.runtime.dispatch(&CapabilityCall::JobEnqueue {
                id: 1,
                job_type: job_type.to_owned(),
                payload: PluginRow::new(),
            });
            assert_eq!(
                result.denial(),
                Some(DenialReason::NotInGrant),
                "{job_type}"
            );
        }
        assert!(fixture.jobs.queued().is_empty());
    }

    // ── Quotas ───────────────────────────────────────────────────────

    #[test]
    fn a_spent_quota_denies_the_call_and_nothing_else() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        let ceiling = fixture.runtime.quotas().kv_reads;
        for id in 0..ceiling {
            let result = fixture.runtime.dispatch(&CapabilityCall::KvGet {
                id: u64::from(id),
                key: "cart".to_owned(),
            });
            assert!(result.denial().is_none(), "call {id}: {result:?}");
        }
        let over = fixture.runtime.dispatch(&CapabilityCall::KvGet {
            id: 999,
            key: "cart".to_owned(),
        });
        assert_eq!(over.denial(), Some(DenialReason::QuotaExceeded));

        // A different capability is untouched: quotas are per surface, so
        // running out of reads does not cost the plugin its ability to enqueue.
        let job = fixture.runtime.dispatch(&CapabilityCall::JobEnqueue {
            id: 1000,
            job_type: "reindex".to_owned(),
            payload: PluginRow::new(),
        });
        assert!(job.denial().is_none(), "{job:?}");
    }

    #[test]
    fn one_plugins_spent_quota_does_not_touch_another_plugins() {
        let manifest = manifest(&everything());
        let store = MemoryKvStore::new();
        let services = CapabilityServices {
            kv: Some(Arc::clone(&store) as Arc<dyn KvStore>),
            ..CapabilityServices::none()
        }
        .for_tenant("alpha");
        let mut first = CapabilityRuntime::new(&manifest, services.clone());
        for id in 0..manifest.quotas.kv_reads {
            let _ = first.dispatch(&CapabilityCall::KvGet {
                id: u64::from(id),
                key: "cart".to_owned(),
            });
        }
        assert_eq!(
            first
                .dispatch(&CapabilityCall::KvGet {
                    id: 999,
                    key: "cart".to_owned()
                })
                .denial(),
            Some(DenialReason::QuotaExceeded)
        );

        let mut second = CapabilityRuntime::new(&manifest, services);
        assert!(
            second
                .dispatch(&CapabilityCall::KvGet {
                    id: 1,
                    key: "cart".to_owned()
                })
                .denial()
                .is_none()
        );
    }

    #[test]
    fn the_shared_call_budget_bounds_the_sum_of_every_surface() {
        let mut manifest = manifest(&everything());
        manifest.quotas.calls = 3;
        let mut runtime = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
                jobs: Some(MemoryJobSink::new() as Arc<dyn JobSink>),
                ..CapabilityServices::none()
            },
        );
        for id in 0..3 {
            assert!(
                runtime
                    .dispatch(&CapabilityCall::KvGet {
                        id,
                        key: "k".to_owned()
                    })
                    .denial()
                    .is_none()
            );
        }
        // The fourth call is refused even though `job_enqueues` is untouched.
        assert_eq!(
            runtime
                .dispatch(&CapabilityCall::JobEnqueue {
                    id: 4,
                    job_type: "reindex".to_owned(),
                    payload: PluginRow::new(),
                })
                .denial(),
            Some(DenialReason::QuotaExceeded)
        );
    }

    #[test]
    fn a_calls_per_second_ceiling_is_shared_across_requests_and_kept_per_capability() {
        let manifest = manifest(&everything());
        // One token per second, not two: the assertion below needs the bucket
        // to still be empty when it runs, and a `per_second` of 2 gave it only
        // 500 ms of slack — enough for a loaded debug-build runner to refill
        // one token between the take and the check and flip this green to red.
        let rate = Arc::new(quota::CapabilityRateLimiter::new(1));
        let services = CapabilityServices {
            kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
            jobs: Some(MemoryJobSink::new() as Arc<dyn JobSink>),
            rate: Some(Arc::clone(&rate)),
            ..CapabilityServices::none()
        };
        let mut first = CapabilityRuntime::new(&manifest, services.clone());
        assert!(
            first
                .dispatch(&CapabilityCall::KvGet {
                    id: 1,
                    key: "k".to_owned()
                })
                .denial()
                .is_none()
        );
        // A *second request* of the same plugin shares the bucket.
        let mut second = CapabilityRuntime::new(&manifest, services);
        assert_eq!(
            second
                .dispatch(&CapabilityCall::KvGet {
                    id: 3,
                    key: "k".to_owned()
                })
                .denial(),
            Some(DenialReason::QuotaExceeded)
        );
        // A different capability has its own bucket, so a busy cache does not
        // cost the plugin its ability to enqueue.
        assert!(
            second
                .dispatch(&CapabilityCall::JobEnqueue {
                    id: 4,
                    job_type: "reindex".to_owned(),
                    payload: PluginRow::new()
                })
                .denial()
                .is_none()
        );
    }

    #[test]
    fn an_oversized_row_or_key_is_refused_before_it_is_stored() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        let long = "k".repeat(MAX_KV_KEY_BYTES + 1);
        assert_eq!(
            fixture
                .runtime
                .dispatch(&CapabilityCall::KvGet { id: 1, key: long })
                .denial(),
            Some(DenialReason::Malformed)
        );
        let wide: PluginRow = (0..=MAX_ROW_COLUMNS)
            .map(|index| (format!("c{index}"), PluginValue::Int(1)))
            .collect();
        assert_eq!(
            fixture
                .runtime
                .dispatch(&CapabilityCall::DbInsert {
                    id: 2,
                    table: "orders".to_owned(),
                    row: wide
                })
                .denial(),
            Some(DenialReason::Malformed)
        );
        assert!(fixture.db.keys().is_empty());
    }

    #[test]
    fn a_missing_backend_is_a_denial_rather_than_a_silent_success() {
        let mut runtime =
            CapabilityRuntime::new(&manifest(&everything()), CapabilityServices::none());
        assert_eq!(
            runtime
                .dispatch(&CapabilityCall::KvSet {
                    id: 1,
                    key: "cart".to_owned(),
                    value: PluginValue::Int(1)
                })
                .denial(),
            Some(DenialReason::Unavailable)
        );
    }

    #[test]
    fn a_response_header_a_plugin_may_not_see_never_reaches_it() {
        // The response side of the outbound allow-list. Nothing exercised it,
        // so deleting the filter left the suite green — and `set-cookie` from
        // an upstream is a cookie the *plugin* then learns.
        let mut fixture = fixture(&everything(), Some("alpha"));
        fixture.http.answer(
            "https://api.example.com/v1",
            OutboundResponse {
                status: 200,
                headers: vec![
                    ("content-type".to_owned(), "application/json".to_owned()),
                    ("set-cookie".to_owned(), "session=stolen".to_owned()),
                    ("x-upstream-secret".to_owned(), "s3cret".to_owned()),
                ],
                body: "{}".to_owned(),
                final_url: "https://api.example.com/v1".to_owned(),
            },
        );
        let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "GET".to_owned(),
            url: "https://api.example.com/v1".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        let CallResult::Ok {
            value: CallValue::Http { headers, .. },
            ..
        } = result
        else {
            panic!("the call should have succeeded: {result:?}")
        };
        assert_eq!(
            headers,
            vec![("content-type".to_owned(), "application/json".to_owned())]
        );
    }

    #[test]
    fn a_method_or_header_count_outside_the_allow_list_is_refused() {
        // A fresh fixture per case, so each refusal is read on its own.
        for method in ["CONNECT", "TRACE", "OPTIONS", " GET", ""] {
            let mut fixture = fixture(&everything(), Some("alpha"));
            let result = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
                id: 1,
                method: method.to_owned(),
                url: "https://api.example.com/".to_owned(),
                headers: Vec::new(),
                body: String::new(),
            });
            assert_eq!(result.denial(), Some(DenialReason::Malformed), "{method}");
            assert!(fixture.http.seen().is_empty(), "{method} left the host");
        }
        let mut fixture = fixture(&everything(), Some("alpha"));
        let many: Vec<(String, String)> = (0..=MAX_OUTBOUND_HEADERS)
            .map(|_| ("accept".to_owned(), "*/*".to_owned()))
            .collect();
        assert_eq!(
            fixture
                .runtime
                .dispatch(&CapabilityCall::HttpFetch {
                    id: 2,
                    method: "GET".to_owned(),
                    url: "https://api.example.com/".to_owned(),
                    headers: many,
                    body: String::new(),
                })
                .denial(),
            Some(DenialReason::Malformed)
        );
        assert!(fixture.http.seen().is_empty());
    }

    #[test]
    fn every_byte_ceiling_a_quota_names_is_actually_applied() {
        // Four ceilings that had no test between them, so four `if` statements
        // nothing reached.
        let mut manifest = manifest(&everything());
        manifest.quotas.kv_value_bytes = 8;
        manifest.quotas.db_rows = 2;
        manifest.quotas.outbound_response_bytes = 8;
        let http = RecordingHttp::new();
        http.answer(
            "https://api.example.com/big",
            OutboundResponse::from_url("https://api.example.com/big", 200, "x".repeat(64)),
        );
        let db = db::MemoryPluginStore::new();
        let mut runtime = CapabilityRuntime::new(
            &manifest,
            CapabilityServices {
                kv: Some(MemoryKvStore::new() as Arc<dyn KvStore>),
                db: Some(Arc::clone(&db) as Arc<dyn PluginStore>),
                http: Some(Arc::clone(&http) as Arc<dyn OutboundHttp>),
                ..CapabilityServices::none()
            }
            .for_tenant("alpha"),
        );

        // `kv_value_bytes`.
        assert_eq!(
            runtime
                .dispatch(&CapabilityCall::KvSet {
                    id: 1,
                    key: "cart".to_owned(),
                    value: PluginValue::Text("far too long a value".to_owned())
                })
                .denial(),
            Some(DenialReason::QuotaExceeded)
        );

        // `db_rows`, both as the clamp on an explicit limit and as the default
        // for `limit: 0`.
        for index in 0..5 {
            let inserted = runtime.dispatch(&CapabilityCall::DbInsert {
                id: 10 + index,
                table: "orders".to_owned(),
                row: row(&[("sku", "A")]),
            });
            assert!(inserted.denial().is_none(), "{inserted:?}");
        }
        for (id, limit) in [(20, 0), (21, 100)] {
            let result = runtime.dispatch(&CapabilityCall::DbQuery {
                id,
                table: "orders".to_owned(),
                filter: PluginRow::new(),
                limit,
                after: None,
            });
            let CallResult::Ok {
                value: CallValue::Rows { rows, .. },
                ..
            } = result
            else {
                panic!("the query should have succeeded: {result:?}")
            };
            assert_eq!(rows.len(), 2, "limit {limit} is clamped to `db_rows`");
        }

        // `outbound_response_bytes`.
        assert_eq!(
            runtime
                .dispatch(&CapabilityCall::HttpFetch {
                    id: 30,
                    method: "GET".to_owned(),
                    url: "https://api.example.com/big".to_owned(),
                    headers: Vec::new(),
                    body: String::new(),
                })
                .denial(),
            Some(DenialReason::ResponseTooLarge)
        );
    }

    #[test]
    fn a_row_id_outside_its_bounds_is_refused_before_the_store_sees_it() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        for row_id in [String::new(), "r".repeat(MAX_ROW_ID_BYTES + 1)] {
            assert_eq!(
                fixture
                    .runtime
                    .dispatch(&CapabilityCall::DbGet {
                        id: 1,
                        table: "orders".to_owned(),
                        row_id: row_id.clone(),
                    })
                    .denial(),
                Some(DenialReason::Malformed),
                "{} bytes",
                row_id.len()
            );
        }
    }

    #[test]
    fn a_full_job_queue_is_a_denial_the_guest_can_read() {
        // `MemoryJobSink::bounded` existed with no caller — test-support API
        // written for a test that was never written, which is the same thing as
        // an untested `BackendError` arm.
        let sink = jobs::MemoryJobSink::bounded(1);
        let mut runtime = CapabilityRuntime::new(
            &manifest(&everything()),
            CapabilityServices {
                jobs: Some(Arc::clone(&sink) as Arc<dyn JobSink>),
                ..CapabilityServices::none()
            }
            .for_tenant("alpha"),
        );
        let enqueue = |id| CapabilityCall::JobEnqueue {
            id,
            job_type: "reindex".to_owned(),
            payload: PluginRow::new(),
        };
        assert!(runtime.dispatch(&enqueue(1)).denial().is_none());
        assert_eq!(
            runtime.dispatch(&enqueue(2)).denial(),
            Some(DenialReason::BackendError)
        );
        assert_eq!(sink.queued().len(), 1);
    }

    #[test]
    fn the_zero_configuration_job_queue_is_finite() {
        // `MemoryJobSink::new()` used to be unbounded, and this slice ships no
        // consumer that removes a queued job — so an application that wired the
        // exported default gave a granted plugin a way to grow the host without
        // limit. The quotas slow that and never stop it. There is no unbounded
        // spelling any more; a default that must be tightened to be safe is a
        // default that will be shipped as it is.
        //
        // Driven against the sink rather than through `dispatch`: the per-request
        // quotas would refuse long before the queue filled, and it is the queue
        // that outlives the request.
        let sink = jobs::MemoryJobSink::new();
        let job = || PluginJob {
            plugin: "shop".to_owned(),
            artifact_sha256: "c".repeat(64),
            job_type: "reindex".to_owned(),
            tenant: Some("alpha".to_owned()),
            payload: PluginRow::new(),
        };
        for index in 0..jobs::DEFAULT_JOB_DEPTH {
            assert!(sink.enqueue(job()).is_ok(), "job {index}");
        }
        let over = sink.enqueue(job());
        assert!(over.is_err(), "{over:?}");
        assert_eq!(sink.queued().len(), jobs::DEFAULT_JOB_DEPTH);
    }

    // ── Audit ────────────────────────────────────────────────────────

    #[test]
    fn one_surface_answers_what_this_plugin_did() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        fixture.http.answer(
            "https://api.example.com/v1",
            OutboundResponse {
                status: 200,
                headers: Vec::new(),
                body: "ok".to_owned(),
                final_url: "https://api.example.com/v1".to_owned(),
            },
        );
        let _ = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 1,
            method: "GET".to_owned(),
            url: "https://api.example.com/v1".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        let _ = fixture.runtime.dispatch(&CapabilityCall::HttpFetch {
            id: 2,
            method: "GET".to_owned(),
            url: "https://attacker.test/".to_owned(),
            headers: Vec::new(),
            body: String::new(),
        });
        let _ = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
            id: 3,
            table: "orders".to_owned(),
            row: row(&[("sku", "A-1")]),
        });
        let _ = fixture.runtime.dispatch(&CapabilityCall::JobEnqueue {
            id: 4,
            job_type: "reindex".to_owned(),
            payload: PluginRow::new(),
        });

        let log = audit::PluginActivityLog::new();
        log.ingest("shop", fixture.runtime.take_events());
        let summary = log.summary("shop", std::time::Duration::from_secs(3600));

        assert_eq!(summary.hosts.get("api.example.com"), Some(&1));
        assert_eq!(summary.tables.get("plugin_shop_orders"), None);
        assert_eq!(summary.tables.get("orders"), Some(&1));
        assert_eq!(summary.job_types.get("reindex"), Some(&1));
        assert_eq!(summary.denied.get("http-fetch"), Some(&1));
        let rendered = summary.to_string();
        assert!(rendered.contains("api.example.com"), "{rendered}");
        assert!(rendered.contains("reindex"), "{rendered}");
    }

    #[test]
    fn the_audit_record_never_carries_a_value() {
        let mut fixture = fixture(&everything(), Some("alpha"));
        let _ = fixture.runtime.dispatch(&CapabilityCall::KvSet {
            id: 1,
            key: "cart".to_owned(),
            value: PluginValue::Text("the customer's card number".to_owned()),
        });
        let _ = fixture.runtime.dispatch(&CapabilityCall::DbInsert {
            id: 2,
            table: "orders".to_owned(),
            row: row(&[("note", "the customer's home address")]),
        });
        let serialized = serde_json::to_string(fixture.runtime.events()).expect("events serialize");
        assert!(!serialized.contains("card number"), "{serialized}");
        assert!(!serialized.contains("home address"), "{serialized}");
        assert!(
            serialized.contains("cart"),
            "the key is a shape, not a value"
        );
    }
}