autumn-web 0.8.0

//! The capability manifest that accompanies a sandboxed plugin artifact.
//!
//! The manifest is the *whole* review surface. An operator who reads it knows
//! everything the plugin may do, because the runtime refuses to do anything the
//! manifest does not name:
//!
//! ```toml
//! name = "autumn-plugin-hello"
//! version = "0.1.0"
//! wire_version = 1
//! prefix = "/hello"
//! capabilities = ["http-request"]
//! sha256 = "…64 hex chars, the module's digest…"
//!
//! [[routes]]
//! method = "GET"
//! path = "/hello/greet"
//!
//! [limits]
//! fuel = 200_000_000
//! memory_bytes = 33_554_432
//! ```
//!
//! # Everything here fails closed
//!
//! Parsing is not "read what you recognise and ignore the rest". An unknown
//! key, an unknown capability name, a future `wire_version`, a declared route
//! outside the declared prefix, a zero or oversized limit, a digest that is not
//! 64 lowercase hex characters — each is a hard error, because every one of
//! them is a case where the operator's reading of the manifest and the
//! runtime's would differ. A manifest an older build cannot fully understand is
//! a manifest it must refuse to run.
//!
//! # The routes are enforced, not documented
//!
//! [`SandboxManifest::routes`] is not advisory metadata. The host builds its
//! router from exactly these `(method, path)` pairs, so a request to an
//! undeclared path under the prefix is a 404 the guest never sees. That is what
//! makes "which routes it mounts" a property of the manifest rather than a
//! promise about the artifact.

use std::fmt;

use serde::{Deserialize, Deserializer, Serialize};

use super::grants::{CapabilityGrants, CapabilityQuotas, ConsentDelta, added};
use crate::route_listing::{RouteClassification, RouteInfo, RouteSource};
use crate::security::path::{dot_segment_len, percent_escapes};

/// The sandbox wire-protocol version this build speaks.
///
/// A manifest declaring any other value is refused at load, so a host and an
/// artifact built from different Autumn versions never guess at each other.
pub const WIRE_VERSION: u32 = 1;

/// Longest accepted plugin name, in bytes.
const MAX_NAME_LEN: usize = 64;

/// Upper bound on a manifest's declared fuel budget.
///
/// Fuel is the CPU ceiling: roughly one unit per executed instruction, so this
/// bounds a runaway guest to seconds of a core, not forever. The ceiling exists
/// so a manifest cannot ask for an *unbounded* budget and call it a limit.
pub const MAX_FUEL: u64 = 100_000_000_000;

/// Upper bound on a manifest's declared linear-memory ceiling (1 GiB).
pub const MAX_MEMORY_BYTES: usize = 1024 * 1024 * 1024;

/// Upper bound on a manifest's declared request-body ceiling (64 MiB).
pub const MAX_REQUEST_BODY_BYTES: usize = 64 * 1024 * 1024;

/// Upper bound on a manifest's declared response ceiling (64 MiB).
pub const MAX_RESPONSE_BYTES: usize = 64 * 1024 * 1024;

/// Upper bound on the memory a plugin may hold across all in-flight requests
/// (1 GiB).
///
/// Bounding the factors separately does not bound the product: 1 GiB × 1024 is
/// two valid factors and a terabyte. And linear memory is not the only thing an
/// in-flight request pins — the buffered request body, the pending stdout frame
/// and the decoded response all live in *host* memory, outside the guest's
/// limiter, so a manifest with a tiny `memory_bytes` and 64 MiB body/response
/// ceilings would pass a memory-only product check and still allocate hundreds
/// of gigabytes. See [`ResourceLimits::request_footprint_bytes`].
pub const MAX_FOOTPRINT_BYTES: u128 = 1024 * 1024 * 1024;

/// Upper bound on a manifest's declared request-body deadline (60 s).
pub const MAX_REQUEST_BODY_TIMEOUT_MS: u64 = 60_000;

/// Upper bound on a manifest's declared concurrency ceiling.
///
/// Each in-flight request holds its own instance, and each instance may hold up
/// to [`ResourceLimits::memory_bytes`], so concurrency × memory is the real
/// host exposure. Bounding it keeps that product reviewable.
pub const MAX_CONCURRENCY: usize = 1024;

/// HTTP methods a declared route may use.
///
/// `CONNECT` and `TRACE` are absent deliberately: neither is a thing a plugin
/// serving its own prefix has any business answering.
const ALLOWED_METHODS: &[&str] = &["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"];

// ── Capabilities ─────────────────────────────────────────────────────────

/// A capability a sandboxed plugin may be granted.
///
/// Every word here binds to a subsystem that *already* enforces scoping —
/// tenancy on the cache and the repositories, named upstreams on the HTTP
/// client, declared types on the job queue — so a grant attaches to an
/// enforcement point rather than creating one. Filesystem access, environment
/// variables, raw SQL, sessions, mail and file storage are not "not granted by
/// default": they do not exist as grantable capabilities at all, so there is no
/// manifest a plugin author can write that asks for them.
///
/// Capabilities are **data, not code**. A plugin granted `db` imports exactly
/// what an ungranted one imports — the guest asks over the same NDJSON channel
/// it already answers on — so growing this vocabulary never widens the module's
/// import surface, and the escape corpus that proves the import surface is
/// closed keeps proving it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize)]
#[serde(rename_all = "kebab-case")]
#[non_exhaustive]
pub enum SandboxCapability {
    /// Serve HTTP requests routed to the plugin's declared prefix.
    HttpRequest,
    /// Read and write a key/value namespace private to (this plugin, this
    /// tenant).
    Kv,
    /// Call the hostnames `[grants].hosts` names, through the framework client.
    HttpOutbound,
    /// Read and write the plugin-owned, tenant-scoped tables `[grants].tables`
    /// names. Never host-application tables, and never raw SQL.
    Db,
    /// Enqueue the job types `[grants].job_types` names.
    Jobs,
    /// Fill the host-declared render slots `[grants].slots` names.
    Render,
}

impl SandboxCapability {
    /// Every capability this build understands, in manifest spelling.
    pub const ALL: &'static [Self] = &[
        Self::HttpRequest,
        Self::Kv,
        Self::HttpOutbound,
        Self::Db,
        Self::Jobs,
        Self::Render,
    ];

    /// The manifest spelling of this capability.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::HttpRequest => "http-request",
            Self::Kv => "kv",
            Self::HttpOutbound => "http-outbound",
            Self::Db => "db",
            Self::Jobs => "jobs",
            Self::Render => "render",
        }
    }

    /// One line an operator can read on a consent screen.
    ///
    /// Each line names the *scope*, not just the surface, because "may use a
    /// database" and "may read and write its own tables, for the active tenant
    /// only" are different approvals and only one of them is true.
    #[must_use]
    pub const fn describe(self) -> &'static str {
        match self {
            Self::HttpRequest => {
                "handle HTTP requests routed to this plugin's own prefix (no other authority)"
            }
            Self::Kv => {
                "read and write a key/value namespace private to this plugin and the active \
                 tenant; no other plugin's or tenant's keys are reachable or visible"
            }
            Self::HttpOutbound => {
                "call the hostnames listed under `[grants].hosts`, and no others, through the \
                 framework's HTTP client"
            }
            Self::Db => {
                "read and write the plugin-owned tables listed under `[grants].tables`, scoped \
                 to the active tenant; no host-application table and no raw SQL"
            }
            Self::Jobs => {
                "enqueue the job types listed under `[grants].job_types`, which run under this \
                 plugin's own grants and quotas"
            }
            Self::Render => {
                "return a fragment for the host-declared render slots listed under \
                 `[grants].slots`; the host renders it, so no script, style or event handler \
                 can cross"
            }
        }
    }

    /// Parse a manifest capability name.
    ///
    /// # Errors
    ///
    /// Returns [`ManifestError::UnknownCapability`] for any name this build
    /// does not understand — an older host must refuse a newer grant, never
    /// silently drop it.
    pub fn parse(raw: &str) -> Result<Self, ManifestError> {
        Self::ALL
            .iter()
            .copied()
            .find(|capability| capability.as_str() == raw)
            .ok_or_else(|| ManifestError::UnknownCapability(rejected(raw)))
    }
}

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

impl<'de> Deserialize<'de> for SandboxCapability {
    fn deserialize<D: Deserializer<'de>>(de: D) -> Result<Self, D::Error> {
        let raw = String::deserialize(de)?;
        Self::parse(&raw).map_err(serde::de::Error::custom)
    }
}

// ── Declared routes ──────────────────────────────────────────────────────

/// One `(method, path)` pair the plugin mounts.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DeclaredRoute {
    /// HTTP method, upper-cased during parsing.
    pub method: String,
    /// Full mounted path, which must be the prefix or live under it.
    pub path: String,
}

// ── Resource limits ──────────────────────────────────────────────────────

/// The most routes a manifest may declare.
///
/// Generous against anything an author writes by hand — a plugin serves a
/// handful of endpoints, not hundreds — and it is the count rather than the
/// bytes that has to be bounded here. A parsed artifact is already capped at
/// `MAX_MANIFEST_BYTES`, but [`SandboxHost::from_module`](crate::plugin_sandbox::SandboxHost::from_module)
/// is public and takes a `SandboxManifest` an embedder built in memory, where
/// no byte ceiling applies.
///
/// Checked before the validation indexes are reserved and scanned, because
/// those scans are linear per route and so quadratic over the list, and because
/// an accepted list is cloned again into the host's owned-route set. Refusing
/// first keeps load-time work bounded by this constant rather than by what a
/// caller chose to hand over.
pub const MAX_ROUTES: usize = 256;

/// Upper bound on a path-shaped manifest field: the plugin's `prefix`, and each
/// declared route's `path`.
///
/// [`MAX_ROUTES`] bounds how many paths a manifest may declare; this bounds how
/// large any one of them may be, and the two together bound the aggregate at
/// 256 KiB. Both ceilings exist for the same reason and neither substitutes for
/// the other: `MAX_MANIFEST_BYTES` bounds a manifest *parsed* out of an
/// artifact, while [`SandboxHost::from_module`](crate::plugin_sandbox::SandboxHost::from_module)
/// is public and takes a `SandboxManifest` whose fields an embedder filled in
/// by hand, where no byte bound applies at all. Without this one, a single
/// route path of a caller's choosing is walked segment by segment, inserted
/// into a `matchit` router, and then cloned again into the host's `OwnedRoutes`
/// for the plugin's lifetime.
///
/// A kilobyte is far past anything an author writes: an axum route template
/// long enough to reach it is already unreadable.
pub const MAX_PATH_LEN: usize = 1024;

// The refusal reasons below name this ceiling in prose, and a `&'static str`
// cannot interpolate a constant. Asserted equal so the two cannot drift.
const _: () = assert!(MAX_PATH_LEN == 1024);

/// The per-request resource ceilings the host enforces for this plugin.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields, default)]
pub struct ResourceLimits {
    /// CPU ceiling for one request, in wasm fuel units (~one per instruction).
    pub fuel: u64,
    /// Linear-memory ceiling for one request's instance, in bytes.
    pub memory_bytes: usize,
    /// Largest request body forwarded into the guest, in bytes. A larger body
    /// is refused with 413 and the guest is never started.
    pub max_request_body_bytes: usize,
    /// Largest response frame accepted from the guest, in bytes.
    pub max_response_bytes: usize,
    /// Largest number of requests this plugin may execute concurrently.
    pub max_concurrency: usize,
    /// How long the host will wait for a request body before giving up, in
    /// milliseconds.
    ///
    /// A request holds its concurrency permit from the moment it is admitted —
    /// that is what makes the footprint below a bound on the whole request and
    /// not just on the part a guest is running. Without a deadline, a client
    /// that dribbles a body could hold a permit indefinitely without ever
    /// starting a guest, and `max_concurrency` such clients would shut the
    /// plugin's prefix with 503s while no sandbox was executing anything.
    pub request_body_timeout_ms: u64,
}

impl Default for ResourceLimits {
    /// Defaults sized for "a plugin renders a page", not for "a plugin does
    /// arbitrary work": generous enough that an honest handler never notices,
    /// small enough that a hostile one is stopped in milliseconds.
    fn default() -> Self {
        Self {
            fuel: 200_000_000,
            memory_bytes: 32 * 1024 * 1024,
            max_request_body_bytes: 1024 * 1024,
            max_response_bytes: 4 * 1024 * 1024,
            max_concurrency: 8,
            request_body_timeout_ms: 5_000,
        }
    }
}

impl ResourceLimits {
    /// The host memory one in-flight request may hold at once.
    ///
    /// Every term is a buffer that exists while a request is being served:
    ///
    /// | Term | What holds it |
    /// | --- | --- |
    /// | `memory_bytes` | the guest instance's linear memory |
    /// | `4 × max_request_body_bytes` | the body is buffered, cloned into the frame, and base64-expanded (≈4/3) into the NDJSON line that becomes the guest's stdin — all live at once |
    /// | `5 × max_response_bytes` | the response side peaks while the answer is *parsed*, not after: the raw NDJSON line is still live (up to `2 ×`), the base64 field may be copied out of it (`~1.34 ×`, when a guest escapes it), and the decoded body is allocated while both are held |
    /// | `8 × MAX_REQUEST_METADATA_BYTES` | the caller's strings, the frame's clone of them, and the *escaped* JSON line — `serde_json` writes a control character as `\u0000`, six bytes for one, so the line alone is up to `6 ×` while the other two are still held |
    /// | table storage | bounded per instance by `MAX_TABLE_ELEMENTS`, at 16 bytes a reference |
    ///
    /// The request terms are deliberately counted at their *simultaneous* peak
    /// rather than at what any one of them costs: a ceiling that assumed the
    /// buffers took turns would be a number nobody could rely on.
    ///
    /// Multiplied by `max_concurrency`, this is what the plugin can cost the
    /// host at any instant — the number a reviewer should actually look at, so
    /// it is the number the validator checks.
    ///
    /// It bounds what the *sandbox* holds while serving: the guest instance,
    /// the buffers on either side of the wire, and the response for as long as
    /// the framework is delivering it — on that last one the permit rides
    /// inside the response buffer, so a slow client extends the slot exactly as
    /// long as it extends the bytes. It does not bound what an embedder chooses
    /// to keep. [`SandboxHost::run`](crate::plugin_sandbox::host::SandboxHost::run)
    /// hands back a `SandboxOutcome` that owns its response, and a caller that
    /// stores a thousand of those has a thousand response bodies resident —
    /// which is a property of that caller's code, not of the plugin's ceiling,
    /// the same as for any other owned value an API returns.
    #[must_use]
    pub const fn request_footprint_bytes(&self) -> u128 {
        (self.memory_bytes as u128)
            // Five, not four, and the fifth is a temporary the term used to miss.
            // At the moment `to_line` runs, four copies of the body are live at
            // once: the caller's, the clone `HostFrame::request` takes, the
            // `String` `BASE64.encode` allocates, and the encoded text
            // `serialize_str` copies into the serializer's output. The last two
            // are 4/3 each, because base64 expands, so the peak is
            // 1 + 1 + 4/3 + 4/3 = 14/3 of the body, over the four this budgeted,
            // by enough to matter at a concurrency near the product's ceiling.
            //
            // Counted rather than removed: serialising the base64 straight into
            // the output would delete the temporary outright, which is the better
            // fix and a larger one — it changes how the frame is written, and
            // proving the allocation is gone needs more than reading the code. The
            // bound is corrected here to what the code does; making the code do
            // less is worth doing on its own.
            .saturating_add((self.max_request_body_bytes as u128).saturating_mul(5))
            .saturating_add((self.max_response_bytes as u128).saturating_mul(5))
            // The instance's tables, bounded by `MAX_TABLE_ELEMENTS` at a
            // generous 16 bytes a reference. Small, but per-instance storage
            // the footprint would otherwise not know about at all.
            .saturating_add(crate::plugin_sandbox::host::MAX_TABLE_ELEMENTS as u128 * 16)
            // The request's metadata: the caller's strings, the frame's clone of
            // them, and the serialised line — the term that was wrong. `4 x`
            // priced the line at the raw byte count, but JSON is an escaping
            // encoding: `serde_json` writes a control character as a six-byte
            // `\uXXXX` escape, and every byte of a metadata field can be one. An
            // HTTP request cannot carry them — the `http` crate refuses control
            // characters in header values and URIs — but `SandboxHost::run` is
            // public and an embedder builds the `SandboxRequest` by hand, so the
            // bound has to hold for the API rather than for the adapter that is
            // merely its politest caller.
            //
            // The ceiling that bounds the raw bytes is the host's rather than this
            // manifest's, but it is per-request storage all the same: leaving it
            // out entirely made this product understate a near-maximum-concurrency
            // plugin by hundreds of megabytes, and pricing it unescaped understated
            // it again by as much.
            .saturating_add(crate::plugin_sandbox::host::MAX_REQUEST_METADATA_BYTES as u128 * 8)
            // The instance's globals, at a generous 16 bytes each. Per-instance
            // storage the footprint would otherwise not know about at all — the
            // same omission the tables term above exists to correct.
            .saturating_add(crate::plugin_sandbox::host::MAX_GLOBALS as u128 * 16)
            // The instance's function entries, at 32 bytes each. Same argument
            // as globals: per-instance storage the product would not otherwise
            // know about.
            .saturating_add(crate::plugin_sandbox::host::MAX_FUNCTIONS as u128 * 32)
            // The host buffers that do not scale with any ceiling this manifest
            // names, and were therefore left to a flat 4 KiB of slack that did
            // not cover them: the 64 KiB stderr budget held for the whole
            // request, the 64 KiB scratch an `fd_write` allocates while that
            // budget is still resident, and the bounded denial ledger. Fixed
            // per request is still per request — at a concurrency near this
            // product's own ceiling they are tens of megabytes.
            .saturating_add(crate::plugin_sandbox::host::FIXED_HOST_BUFFER_BYTES as u128)
    }

    /// Every limit, as `(field, value)`, in declaration order.
    ///
    /// One list, read by validation and by the upgrade diff alike — the failure
    /// mode of writing them out twice is a ceiling that is enforced but that an
    /// upgrade can raise without anyone being asked.
    #[must_use]
    pub const fn fields(&self) -> [(&'static str, u128); 6] {
        [
            ("fuel", self.fuel as u128),
            ("memory_bytes", self.memory_bytes as u128),
            (
                "max_request_body_bytes",
                self.max_request_body_bytes as u128,
            ),
            ("max_response_bytes", self.max_response_bytes as u128),
            ("max_concurrency", self.max_concurrency as u128),
            (
                "request_body_timeout_ms",
                self.request_body_timeout_ms as u128,
            ),
        ]
    }

    fn validate(&self) -> Result<(), ManifestError> {
        let checks: [(&str, u128, u128); 6] = [
            ("fuel", u128::from(self.fuel), u128::from(MAX_FUEL)),
            (
                "memory_bytes",
                self.memory_bytes as u128,
                MAX_MEMORY_BYTES as u128,
            ),
            (
                "max_request_body_bytes",
                self.max_request_body_bytes as u128,
                MAX_REQUEST_BODY_BYTES as u128,
            ),
            (
                "max_response_bytes",
                self.max_response_bytes as u128,
                MAX_RESPONSE_BYTES as u128,
            ),
            (
                "max_concurrency",
                self.max_concurrency as u128,
                MAX_CONCURRENCY as u128,
            ),
            (
                "request_body_timeout_ms",
                u128::from(self.request_body_timeout_ms),
                u128::from(MAX_REQUEST_BODY_TIMEOUT_MS),
            ),
        ];
        for (field, value, max) in checks {
            // Zero is refused as well as oversized: a zero ceiling is not "no
            // limit" but "cannot run", and a manifest that says it by accident
            // should say so at load rather than at the first request.
            if value == 0 || value > max {
                return Err(ManifestError::LimitOutOfRange { field, value, max });
            }
        }
        let footprint = self
            .request_footprint_bytes()
            .saturating_mul(self.max_concurrency as u128);
        if footprint > MAX_FOOTPRINT_BYTES {
            return Err(ManifestError::LimitOutOfRange {
                field: "the per-request host footprint × max_concurrency",
                value: footprint,
                max: MAX_FOOTPRINT_BYTES,
            });
        }
        Ok(())
    }
}

// ── Errors ───────────────────────────────────────────────────────────────

/// Why a manifest was refused.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum ManifestError {
    /// The TOML did not parse, or carried a key this build does not know.
    Toml(String),
    /// The manifest could not be rendered back to TOML.
    Serialize(String),
    /// A capability name this build does not understand.
    UnknownCapability(String),
    /// The manifest declares a wire version this build does not speak.
    UnsupportedWireVersion {
        /// The version the manifest declared.
        found: u32,
        /// The version this build speaks.
        supported: u32,
    },
    /// The plugin name is empty, over-long, or carries characters that have no
    /// business in a log line or a path.
    InvalidName(String),
    /// The declared version string is empty or over-long.
    InvalidVersion(String),
    /// The route prefix is not a plain, absolute, single-or-multi-segment path.
    InvalidPrefix {
        /// The offending prefix.
        prefix: String,
        /// Why it was refused.
        reason: &'static str,
    },
    /// The manifest grants no capability that would let the plugin serve.
    MissingCapability(SandboxCapability),
    /// The manifest declares no routes, so it could never serve anything.
    NoRoutes,
    /// The manifest declares more than [`MAX_ROUTES`] routes.
    TooManyRoutes {
        /// How many it declared.
        found: usize,
        /// The ceiling.
        max: usize,
    },
    /// A declared route uses a method the sandbox will not mount.
    InvalidMethod(String),
    /// A declared route's path is malformed.
    InvalidRoutePath {
        /// The offending path.
        path: String,
        /// Why it was refused.
        reason: &'static str,
    },
    /// A declared route does not live under the declared prefix.
    RouteOutsidePrefix {
        /// The offending route's method.
        method: String,
        /// The offending route's path.
        path: String,
        /// The prefix it was measured against.
        prefix: String,
    },
    /// Two declared routes are one route as far as the router is concerned.
    ConflictingRoutes {
        /// The route declared first.
        first: String,
        /// The route that collided with it.
        second: String,
    },
    /// The same `(method, path)` pair is declared twice.
    DuplicateRoute {
        /// The duplicated method.
        method: String,
        /// The duplicated path.
        path: String,
    },
    /// The same capability is granted twice.
    DuplicateCapability(SandboxCapability),
    /// The module digest is not 64 lowercase hex characters.
    InvalidDigest(String),
    /// A declared limit is zero or above this build's ceiling.
    LimitOutOfRange {
        /// Which limit.
        field: &'static str,
        /// The declared value.
        value: u128,
        /// This build's ceiling.
        max: u128,
    },
    /// A `[grants]` list names things a capability the manifest never asked for
    /// would be needed to reach.
    GrantWithoutCapability {
        /// The capability the list belongs to.
        capability: SandboxCapability,
        /// The `[grants]` field.
        field: &'static str,
    },
    /// A capability was granted but its `[grants]` list is empty, so it names
    /// nothing it could ever act on.
    CapabilityWithoutGrant {
        /// The capability with nothing to act on.
        capability: SandboxCapability,
        /// The `[grants]` field that should have named something.
        field: &'static str,
    },
    /// A `[grants]` entry is not shaped like the thing a physical name is
    /// derived from.
    InvalidGrantEntry {
        /// The `[grants]` field.
        field: &'static str,
        /// The offending entry.
        entry: String,
        /// What was expected instead.
        reason: &'static str,
    },
    /// The same `[grants]` entry appears twice in one list.
    DuplicateGrantEntry {
        /// The `[grants]` field.
        field: &'static str,
        /// The repeated entry.
        entry: String,
    },
    /// A `[grants]` list is longer than this build will hold.
    TooManyGrantEntries {
        /// The `[grants]` field.
        field: &'static str,
        /// How many it declared.
        found: usize,
        /// The ceiling.
        max: usize,
    },
    /// A declared quota is zero or above this build's ceiling.
    QuotaOutOfRange {
        /// Which quota.
        field: &'static str,
        /// The declared value.
        value: u32,
        /// This build's ceiling.
        max: u32,
    },
}

impl fmt::Display for ManifestError {
    #[allow(
        clippy::too_many_lines,
        reason = "one arm per refusal, each writing the sentence an operator reads; the length \
                  is the vocabulary's, and splitting it would separate a variant from its words"
    )]
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Toml(detail) => write!(f, "malformed sandbox plugin manifest: {detail}"),
            Self::Serialize(detail) => {
                write!(f, "could not render the sandbox plugin manifest: {detail}")
            }
            Self::UnknownCapability(name) => write!(
                f,
                "unknown sandbox capability `{name}`; this build understands: {known}. \
                 A capability this host cannot enforce is refused rather than ignored",
                known = SandboxCapability::ALL
                    .iter()
                    .map(|cap| cap.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            ),
            Self::UnsupportedWireVersion { found, supported } => write!(
                f,
                "sandbox plugin manifest declares wire_version {found}, but this build speaks \
                 version {supported}"
            ),
            Self::InvalidName(name) => write!(
                f,
                "invalid sandbox plugin name {name:?}; expected 1..={MAX_NAME_LEN} characters of \
                 `a-z A-Z 0-9 - _ .`"
            ),
            Self::InvalidVersion(version) => {
                write!(f, "invalid sandbox plugin version {version:?}")
            }
            Self::InvalidPrefix { prefix, reason } => {
                write!(f, "invalid sandbox plugin prefix {prefix:?}: {reason}")
            }
            Self::MissingCapability(cap) => write!(
                f,
                "the manifest grants no `{cap}` capability, so the plugin could never serve a \
                 request; add it to `capabilities` or do not install this plugin"
            ),
            Self::NoRoutes => write!(
                f,
                "the manifest declares no `[[routes]]`; a sandboxed plugin serves exactly the \
                 routes it declares, so an empty list can never serve anything"
            ),
            Self::TooManyRoutes { found, max } => write!(
                f,
                "the manifest declares {found} routes, over the {max}-route ceiling; validating \
                 a route list scans the routes already seen, so the work grows with the square \
                 of the list, and the list is held again for the lifetime of the plugin"
            ),
            Self::InvalidMethod(method) => write!(
                f,
                "invalid sandbox plugin route method {method:?}; expected one of {allowed}",
                allowed = ALLOWED_METHODS.join(", ")
            ),
            Self::InvalidRoutePath { path, reason } => {
                write!(f, "invalid sandbox plugin route path {path:?}: {reason}")
            }
            Self::RouteOutsidePrefix {
                method,
                path,
                prefix,
            } => write!(
                f,
                "declared route `{method} {path}` is outside the declared prefix `{prefix}`; a \
                 sandboxed plugin may only mount under its own prefix"
            ),
            Self::ConflictingRoutes { first, second } => write!(
                f,
                "declared routes `{first}` and `{second}` are the same route to the router, so \
                 mounting both is impossible; give them distinct paths"
            ),
            Self::DuplicateRoute { method, path } => {
                write!(f, "declared route `{method} {path}` appears twice")
            }
            Self::DuplicateCapability(cap) => write!(
                f,
                "the manifest grants `{cap}` more than once; a repeated grant conveys no \
                 additional authority, so list each capability exactly once",
                cap = cap.as_str()
            ),
            Self::InvalidDigest(digest) => write!(
                f,
                "invalid module digest {digest:?}; expected 64 lowercase hex characters"
            ),
            Self::LimitOutOfRange { field, value, max } => write!(
                f,
                "sandbox limit `{field}` = {value} is out of range; expected 1..={max}"
            ),
            Self::GrantWithoutCapability { capability, field } => write!(
                f,
                "`[grants].{field}` names something only the `{capability}` capability could \
                 reach, but `capabilities` does not ask for it; the consent screen and the \
                 runtime would disagree about what this plugin may do"
            ),
            Self::CapabilityWithoutGrant { capability, field } => write!(
                f,
                "the manifest grants `{capability}` but `[grants].{field}` is empty, so the \
                 capability names nothing it could ever act on; an operator who approved it \
                 would have approved authority the runtime can never honour"
            ),
            Self::InvalidGrantEntry {
                field,
                entry,
                reason,
            } => write!(f, "invalid `[grants].{field}` entry {entry:?}: {reason}"),
            Self::DuplicateGrantEntry { field, entry } => write!(
                f,
                "`[grants].{field}` names {entry:?} more than once; a repeat conveys no \
                 authority the first entry did not"
            ),
            Self::TooManyGrantEntries { field, found, max } => write!(
                f,
                "`[grants].{field}` declares {found} entries, over the {max}-entry ceiling; the \
                 list is scanned against itself during validation and consulted on every call"
            ),
            Self::QuotaOutOfRange { field, value, max } => write!(
                f,
                "sandbox quota `{field}` = {value} is out of range; expected 1..={max}"
            ),
        }
    }
}

impl std::error::Error for ManifestError {}

// ── The manifest ─────────────────────────────────────────────────────────

/// A sandboxed plugin's capability manifest.
///
/// Construct one with [`SandboxManifest::parse`]; the constructor validates, so
/// a `SandboxManifest` value in hand is always one this build is willing to
/// enforce.
///
/// Deliberately **not** `#[non_exhaustive]`, unlike most of this subsystem's
/// types: building one by hand is a supported path and one this repository
/// depends on — `tests/sandbox_manifest_seal_alloc_gate.rs` hands `seal` a
/// manifest with 20,000 routes precisely because `parse` would refuse it, and
/// there is no other way to reach the code under test. That is also why
/// [`SandboxHost::from_module`](crate::plugin_sandbox::SandboxHost::from_module)
/// re-validates rather than trusting that parsing ran.
///
/// The cost is that adding a field to this struct is a source break for anyone
/// writing such a literal, and the vocabulary is expected to grow. Prefer
/// `parse` and edit the public fields of what comes back; the slice that adds
/// the next capability will say so in the changelog.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SandboxManifest {
    /// The plugin's name, used for route attribution (`plugin:<name>`), for
    /// duplicate-registration detection, and in every log line.
    pub name: String,
    /// The plugin's own version string, shown to the operator.
    pub version: String,
    /// The sandbox wire version the artifact speaks. Must equal
    /// [`WIRE_VERSION`].
    pub wire_version: u32,
    /// The single URL prefix under which this plugin mounts.
    pub prefix: String,
    /// The capabilities the plugin requires.
    pub capabilities: Vec<SandboxCapability>,
    /// Lowercase hex SHA-256 of the wasm module the manifest describes.
    pub sha256: String,
    /// The routes the plugin mounts. The host's router is built from exactly
    /// this list.
    #[serde(default)]
    pub routes: Vec<DeclaredRoute>,
    /// Per-request resource ceilings.
    #[serde(default)]
    pub limits: ResourceLimits,
    /// The named things each granted capability is scoped to (issue #1632).
    ///
    /// Empty for a plugin that only serves its own prefix, which is what a
    /// first-slice manifest looks like and why this defaults rather than being
    /// required.
    #[serde(default)]
    pub grants: CapabilityGrants,
    /// Per-request ceilings on each capability, operator-configurable.
    #[serde(default)]
    pub quotas: CapabilityQuotas,
}

impl SandboxManifest {
    /// Parse and validate a manifest from TOML.
    ///
    /// # Errors
    ///
    /// Returns a [`ManifestError`] for anything this build cannot fully
    /// understand or enforce — see the module documentation for why every such
    /// case is an error rather than a warning.
    pub fn parse(toml_src: &str) -> Result<Self, ManifestError> {
        let mut manifest: Self =
            toml::from_str(toml_src).map_err(|err| ManifestError::Toml(err.to_string()))?;
        for route in &mut manifest.routes {
            route.method = route.method.to_ascii_uppercase();
        }
        manifest.validate()?;
        Ok(manifest)
    }

    /// Render the manifest back to TOML.
    ///
    /// # Errors
    ///
    /// Returns [`ManifestError::Serialize`] if the value cannot be serialized.
    pub fn to_toml(&self) -> Result<String, ManifestError> {
        toml::to_string_pretty(self).map_err(|err| ManifestError::Serialize(err.to_string()))
    }

    /// Whether this manifest grants `capability`.
    ///
    /// Named `is_granted` rather than `grants` because `grants` is now the
    /// *field* holding what each capability is scoped to, and
    /// `manifest.grants(Render)` two lines above `manifest.grants.slots` is a
    /// thing a reader has to stop and parse.
    #[must_use]
    pub fn is_granted(&self, capability: SandboxCapability) -> bool {
        self.capabilities.contains(&capability)
    }

    /// The route metadata to hand to
    /// [`AppBuilder::declare_plugin_routes`](crate::app::AppBuilder::declare_plugin_routes).
    ///
    /// Routes are classified [`Public`](RouteClassification::Public) because
    /// that is the truth about them: the sandbox grants no session, auth or
    /// database capability, so a sandboxed route is unauthenticated *by
    /// construction* and could not be otherwise. Leaving them unclassified
    /// would fail `autumn routes audit` for a posture that is proven, not
    /// unproven.
    #[must_use]
    pub fn route_infos(&self) -> Vec<RouteInfo> {
        let mut infos = Vec::with_capacity(self.routes.len());
        for route in &self.routes {
            let info = |method: &str| RouteInfo {
                method: method.to_owned(),
                path: route.path.clone(),
                handler: format!("sandbox:{}", self.name),
                source: RouteSource::Plugin(self.name.clone()),
                middleware: vec!["sandboxed".to_owned()],
                classification: RouteClassification::Public,
                ..RouteInfo::default()
            };
            infos.push(info(&route.method));
            // HTTP defines HEAD as GET without the body, and axum's method
            // router dispatches a HEAD with no HEAD route to the GET one. That
            // is correct behaviour, but it means a manifest listing only GET
            // serves a method its own consent screen never named — so the
            // implication is reported rather than left implicit.
            if route.method == "GET" && !self.declares("HEAD", &route.path) {
                infos.push(info("HEAD"));
            }
        }
        infos
    }

    /// Whether the manifest declares this exact `(method, path)` pair.
    fn declares(&self, method: &str, path: &str) -> bool {
        self.routes
            .iter()
            .any(|route| route.method == method && route.path == path)
    }

    /// The operator-facing consent screen: what this plugin may do, what it
    /// may not, and which bytes were reviewed.
    #[must_use]
    pub fn consent_summary(&self) -> String {
        use std::fmt::Write as _;

        let mut out = String::new();
        // `write!` to a `String` is infallible; the results are dropped rather
        // than unwrapped so this stays panic-free by construction.
        let _ = writeln!(
            out,
            "Sandboxed plugin: {name} {version}",
            name = self.name,
            version = self.version
        );
        let _ = writeln!(out, "  module sha256: {}", self.sha256);
        let _ = writeln!(out, "  mounts prefix: {}", self.prefix);
        out.push_str("  routes it serves (and only these):\n");
        for route in &self.routes {
            let _ = writeln!(out, "    {} {}", route.method, route.path);
            if route.method == "GET" && !self.declares("HEAD", &route.path) {
                let _ = writeln!(
                    out,
                    "    HEAD {} (HTTP serves HEAD wherever it serves GET)",
                    route.path
                );
            }
        }
        out.push_str("  capabilities granted:\n");
        for capability in &self.capabilities {
            let _ = writeln!(
                out,
                "    {name} — {describe}",
                name = capability.as_str(),
                describe = capability.describe()
            );
        }
        // The scope of each grant, right under the capability list, because a
        // capability name answers "may it?" and nothing in it answers "to
        // what?". An operator reading `http-outbound` alone has approved the
        // open internet.
        for (label, capability) in [
            (
                "outbound hosts it may call",
                SandboxCapability::HttpOutbound,
            ),
            ("tenant-scoped tables it owns", SandboxCapability::Db),
            ("job types it may enqueue", SandboxCapability::Jobs),
            ("render slots it may fill", SandboxCapability::Render),
        ] {
            let Some(entries) = self
                .grants
                .list_for(capability)
                .filter(|list| !list.is_empty())
            else {
                continue;
            };
            let _ = writeln!(out, "  {label} (and only these):");
            for entry in entries {
                let _ = writeln!(out, "    {entry}");
            }
        }
        // Only the authority this build cannot grant at all. Printing "no
        // database access" under a manifest that was just granted `db` is a
        // consent screen contradicting itself, and the reader believes the
        // reassuring half.
        out.push_str("  denied, with no way to ask for it in this version:\n    ");
        let mut ungranted: Vec<&'static str> = vec![
            "filesystem access",
            "environment variables",
            "session, auth and credential access",
            "mail and file storage",
            "raw SQL and host-application tables",
        ];
        if !self.is_granted(SandboxCapability::HttpOutbound) {
            ungranted.insert(1, "outbound network access");
        }
        if !self.is_granted(SandboxCapability::Db) {
            ungranted.insert(1, "database access");
        }
        if !self.is_granted(SandboxCapability::Kv) {
            ungranted.insert(1, "key/value storage");
        }
        if !self.is_granted(SandboxCapability::Jobs) {
            ungranted.insert(1, "background jobs");
        }
        out.push_str(&ungranted.join(", "));
        out.push_str(",\n    and any host authority not listed above\n");
        out.push_str("  per-request capability quotas:\n");
        for (field, value) in self.quotas.fields() {
            let _ = writeln!(out, "    {field} = {value}");
        }
        out.push_str("  resource ceilings per request:\n");
        let _ = writeln!(
            out,
            "    cpu {fuel} fuel units, memory {memory} bytes, request body {body} bytes\n    \
             (read within {body_ms} ms), response {response} bytes, at most {concurrency} \
             concurrent requests",
            fuel = self.limits.fuel,
            memory = self.limits.memory_bytes,
            body = self.limits.max_request_body_bytes,
            response = self.limits.max_response_bytes,
            body_ms = self.limits.request_body_timeout_ms,
            concurrency = self.limits.max_concurrency,
        );
        out
    }

    /// Everything this manifest asks for that `previous` — the manifest the
    /// operator already approved — did not.
    ///
    /// The install flow asks this, not a version comparison, whether to
    /// re-prompt. A plugin's authority can only grow in a manifest, and a
    /// version string is the author's to write.
    ///
    /// Growth only: dropping a capability, a host or a table, or lowering a
    /// quota, asks for less than was already approved. Prompting for those
    /// trains operators to click through the prompt that matters.
    #[must_use]
    pub fn consent_delta_from(&self, previous: &Self) -> ConsentDelta {
        ConsentDelta {
            added_capabilities: added(&previous.capabilities, &self.capabilities),
            added_hosts: added(&previous.grants.hosts, &self.grants.hosts),
            added_tables: added(&previous.grants.tables, &self.grants.tables),
            added_job_types: added(&previous.grants.job_types, &self.grants.job_types),
            added_slots: added(&previous.grants.slots, &self.grants.slots),
            raised_quotas: previous
                .quotas
                .fields()
                .into_iter()
                .zip(self.quotas.fields())
                .filter(|((field, _), _)| {
                    // A quota bounding a capability this upgrade no longer asks
                    // for is not new authority: the calls it bounds cannot be
                    // made at all. Reporting it made a *narrowing* upgrade —
                    // dropping `kv` while leaving a raised `kv_reads` behind —
                    // exit non-zero from `plugin inspect --against`, which is
                    // exactly the prompt this delta exists to avoid, on the
                    // change least in need of one.
                    super::grants::CapabilityQuotas::governed_by(field)
                        .is_none_or(|capability| self.is_granted(capability))
                })
                .filter_map(|((field, approved), (_, requested))| {
                    (requested > approved).then_some((field, approved, requested))
                })
                .collect(),
            // Routes are the router, not a description of it — an added one is
            // an endpoint the approved manifest did not serve. Compared as the
            // consent screen prints them, including the HEAD a declared GET
            // implies, so a GET added to an existing path is caught with it.
            added_routes: added(&previous.consent_routes(), &self.consent_routes()),
            raised_limits: previous
                .limits
                .fields()
                .into_iter()
                .zip(self.limits.fields())
                .filter_map(|((field, approved), (_, requested))| {
                    (requested > approved).then_some((field, approved, requested))
                })
                .collect(),
        }
    }

    /// The routes the consent screen names, as `"METHOD /path"`.
    ///
    /// Built from [`route_infos`](Self::route_infos) rather than from
    /// `self.routes`, so the HEAD that HTTP serves wherever it serves GET is in
    /// the list an upgrade is diffed against. A manifest that adds a bare GET
    /// adds two mounted routes, and both are authority.
    #[must_use]
    fn consent_routes(&self) -> Vec<String> {
        self.route_infos()
            .into_iter()
            .map(|route| format!("{} {}", route.method, route.path))
            .collect()
    }

    pub(crate) fn validate(&self) -> Result<(), ManifestError> {
        if self.wire_version != WIRE_VERSION {
            return Err(ManifestError::UnsupportedWireVersion {
                found: self.wire_version,
                supported: WIRE_VERSION,
            });
        }
        validate_name(&self.name)?;
        // `version` is rendered verbatim on the consent screen an operator reads
        // before agreeing to run the artifact. A free-form field there can
        // rewrite the lines above it with terminal escapes — hide a route, hide
        // a capability, forge a verdict — so it gets the same treatment as the
        // name: printable ASCII, no spaces, bounded.
        let version_ok = !self.version.is_empty()
            && self.version.len() <= MAX_NAME_LEN
            && self.version.chars().all(|ch| ch.is_ascii_graphic());
        if !version_ok {
            return Err(ManifestError::InvalidVersion(rejected(&self.version)));
        }
        validate_prefix(&self.prefix)?;
        if !self.is_granted(SandboxCapability::HttpRequest) {
            return Err(ManifestError::MissingCapability(
                SandboxCapability::HttpRequest,
            ));
        }
        // A repeat grants nothing the first did, so no manifest is made legal by one —
        // and every request clones this vector and serialises it into the frame. Left
        // unbounded it is per-request work `request_footprint_bytes` never counted and
        // `encoding_fuel` never priced, bought once in a manifest and paid for on every
        // call. Refusing here, rather than deduplicating, keeps the list an operator reads
        // on the consent screen the same list the guest is handed.
        //
        // Scanned in place rather than into a `Vec::with_capacity`: that vector was only
        // ever a duplicate set, dropped below without being read, so sizing it from
        // `self.capabilities.len()` performed — on a list a direct caller chose — exactly
        // the unbounded allocation this check exists to refuse.
        //
        // Pigeonhole bounds the scan: a list longer than the number of distinct
        // capabilities must repeat one, so a duplicate is certain within the first
        // `ALL.len() + 1` entries and looking past them cannot change the answer. Derived
        // from `ALL` rather than written as a number, so it stays right as the vocabulary
        // grows.
        //
        // Sized from the vocabulary, never from `self.capabilities.len()`. That
        // length is the caller's to choose — `from_module` takes a manifest an
        // embedder filled in by hand, where no byte ceiling applies — and this
        // type is no longer the ZST it was when it had one variant, so a
        // capacity taken from it is an allocation an untrusted number sizes.
        // Pigeonhole is what makes the smaller capacity sufficient rather than
        // merely cheaper: a list longer than the number of distinct
        // capabilities must repeat one, so the loop below returns before it
        // could ever push entry `ALL.len() + 1`.
        let mut granted: Vec<SandboxCapability> =
            Vec::with_capacity(SandboxCapability::ALL.len().saturating_add(1));
        for capability in &self.capabilities {
            if granted.contains(capability) {
                return Err(ManifestError::DuplicateCapability(*capability));
            }
            granted.push(*capability);
        }
        // After the duplicate scan, so `granted` is the deduplicated list, and
        // before the routes, so a manifest whose grant table contradicts its
        // capability list is refused on the contradiction rather than on
        // whichever route happens to be malformed too.
        self.grants.validate(&granted)?;
        self.quotas.validate()?;
        // A granted table whose physical name cannot be derived — because the
        // plugin's name and the table's together overrun the identifier ceiling
        // — is authority the consent screen displays and every call then denies
        // as `malformed`. Refusing at load is the difference between an
        // operator learning at install and an author learning at 3am: the same
        // argument `FuelBelowFixedCharges` rests on.
        for table in &self.grants.tables {
            if !crate::plugin_sandbox::capability::db::is_derivable(&self.name, table) {
                return Err(ManifestError::InvalidGrantEntry {
                    field: "tables",
                    entry: rejected(table),
                    reason: "no physical table name can be derived for this plugin name and \
                             table together; both are escaped into one identifier, and the \
                             result must fit in 63 bytes",
                });
            }
        }
        validate_digest(&self.sha256)?;
        if self.routes.is_empty() {
            return Err(ManifestError::NoRoutes);
        }
        // Before the indexes below are reserved or scanned: both are linear per
        // route and so quadratic over the list, and `with_capacity` would size
        // them from a number the caller chose.
        if self.routes.len() > MAX_ROUTES {
            return Err(ManifestError::TooManyRoutes {
                found: self.routes.len(),
                max: MAX_ROUTES,
            });
        }
        let mut seen: Vec<(&str, &str)> = Vec::with_capacity(self.routes.len());
        // The same engine the mount will use, so "these two are one route" is
        // decided here rather than discovered by a panic at boot. Two routes
        // that differ only by method share one template legitimately, so a path
        // already inserted is skipped instead of self-conflicting.
        let mut shapes: matchit::Router<()> = matchit::Router::new();
        let mut inserted: Vec<&str> = Vec::with_capacity(self.routes.len());
        for route in &self.routes {
            if !ALLOWED_METHODS.contains(&route.method.as_str()) {
                return Err(ManifestError::InvalidMethod(rejected(&route.method)));
            }
            validate_route_path(&route.path)?;
            if !path_is_under_prefix(&route.path, &self.prefix) {
                return Err(ManifestError::RouteOutsidePrefix {
                    method: rejected(&route.method),
                    path: rejected(&route.path),
                    prefix: rejected(&self.prefix),
                });
            }
            let key = (route.method.as_str(), route.path.as_str());
            if seen.contains(&key) {
                return Err(ManifestError::DuplicateRoute {
                    method: rejected(&route.method),
                    path: rejected(&route.path),
                });
            }
            seen.push(key);

            if !inserted.contains(&route.path.as_str()) {
                if let Err(matchit::InsertError::Conflict { with }) =
                    shapes.insert(route.path.as_str(), ())
                {
                    return Err(ManifestError::ConflictingRoutes {
                        first: rejected(&with),
                        second: rejected(&route.path),
                    });
                }
                inserted.push(route.path.as_str());
            }
        }
        self.limits.validate()
    }
}

/// Whether `path` is the prefix itself or a path beneath it.
///
/// The `/` check is what stops `/helloworld` from passing as "under `/hello`" —
/// a string-prefix test would mount a plugin over a sibling route's namespace.
fn path_is_under_prefix(path: &str, prefix: &str) -> bool {
    path == prefix
        || path
            .strip_prefix(prefix)
            .is_some_and(|rest| rest.starts_with('/'))
}

fn validate_name(name: &str) -> Result<(), ManifestError> {
    let legal = !name.is_empty()
        && name.len() <= MAX_NAME_LEN
        && name
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.'))
        && name != "."
        && name != "..";
    if legal {
        Ok(())
    } else {
        Err(ManifestError::InvalidName(rejected(name)))
    }
}

fn validate_digest(digest: &str) -> Result<(), ManifestError> {
    let legal = digest.len() == 64
        && digest
            .chars()
            .all(|ch| ch.is_ascii_digit() || ('a'..='f').contains(&ch));
    if legal {
        Ok(())
    } else {
        Err(ManifestError::InvalidDigest(rejected(digest)))
    }
}

/// A prefix must be a plain, absolute path with at least one real segment and
/// no routing syntax: it is a containment boundary, and a boundary that can
/// match dynamically is not one.
/// Characters whose whole purpose is to change how the text around them is
/// displayed, rather than what it says.
///
/// Bidi overrides and isolates reorder a run; the zero-width and invisible
/// characters hide a boundary. `char::is_control` catches neither, because
/// these are formatting characters rather than control codes.
///
/// Used by two surfaces an operator reads and a guest influences: route-path
/// validation, which *refuses* them because a path must mean one thing; and
/// [`guest_text`](super::host::guest_text), which *escapes* them because a
/// failure detail is evidence and has to survive to be read.
pub(super) const fn is_display_reordering(ch: char) -> bool {
    matches!(ch,
        // Unicode's format class (general category Cf) that is *not* also
        // default-ignorable. Enumerated rather than queried because the
        // standard library exposes no general-category predicate — `is_control`
        // answers for Cc only, and `is_whitespace` for the separators, so a
        // format character passes both.
        '\u{0600}'..='\u{0605}'            // Arabic number signs
        | '\u{06DD}'                        // Arabic end of ayah
        | '\u{070F}'                        // Syriac abbreviation mark
        | '\u{0890}'..='\u{0891}'          // Arabic pound and piastre marks
        | '\u{08E2}'                        // Arabic disputed end of ayah
        | '\u{110BD}'                       // Kaithi number sign
        | '\u{110CD}'                       // Kaithi number sign above
        | '\u{13430}'..='\u{1343F}'        // Egyptian hieroglyph formats
        | '\u{FFF9}'..='\u{FFFB}'          // interlinear annotation

        // Unicode's `Default_Ignorable_Code_Point` set, in full. This is the
        // property that actually names the hazard — "a renderer may show
        // nothing here" — and it is not the same set as Cf. Enumerating Cf and
        // adding exceptions was the shape of two bugs in a row: first U+00AD
        // SOFT HYPHEN, then U+115F HANGUL CHOSEONG FILLER, which is general
        // category Lo. A *letter*, neither whitespace nor control nor format,
        // that renders as nothing — so `/hello/ad\u{115F}min` reads as
        // `/hello/admin` on the consent screen while the router mounts a path
        // that is not it. The Hangul fillers below are the same trick four
        // times over.
        //
        // Ranges that are partly unassigned (U+2065, U+FFF0..U+FFF8, most of
        // the U+E0000 block) are covered deliberately: the property is stable
        // for them, and a code point reserved to be invisible is exactly what a
        // future spoof would reach for.
        | '\u{00AD}'                        // soft hyphen
        | '\u{034F}'                        // combining grapheme joiner
        | '\u{061C}'                        // Arabic letter mark
        | '\u{115F}'..='\u{1160}'          // Hangul choseong/jungseong filler
        | '\u{17B4}'..='\u{17B5}'          // Khmer inherent vowels
        | '\u{180B}'..='\u{180F}'          // Mongolian variation and vowel separators
        | '\u{200B}'..='\u{200F}'          // zero-width, LRM/RLM
        | '\u{202A}'..='\u{202E}'          // bidi embedding and override
        | '\u{2060}'..='\u{206F}'          // word joiner, invisibles, isolates, deprecated
        | '\u{3164}'                        // Hangul filler
        | '\u{FE00}'..='\u{FE0F}'          // variation selectors 1-16
        | '\u{FEFF}'                        // zero-width no-break space
        | '\u{FFA0}'                        // halfwidth Hangul filler
        | '\u{FFF0}'..='\u{FFF8}'          // reserved, default-ignorable
        | '\u{1BCA0}'..='\u{1BCA3}'        // Duployan shorthand formats
        | '\u{1D173}'..='\u{1D17A}'        // musical notation formats
        | '\u{E0000}'..='\u{E0FFF}'        // tags and variation selectors 17-256
    )
}

/// Carry a rejected value into its error as a bounded, neutralised excerpt.
///
/// Every caller of this rejects a value for its *shape* — too long, not 64 hex
/// digits, not a known capability — which is decided from the value without
/// ever needing a second copy of it. Cloning it wholesale to say so inverts
/// that: `SandboxManifest`'s fields are public, so a direct
/// [`SandboxHost::from_module`](super::host::SandboxHost::from_module) caller
/// can hand over a name of any size, and the rejection would then hold the
/// original, this copy, and the `to_string()` `from_module` formats out of it —
/// three of the thing whose size was the objection. The container reader's
/// `MAX_MANIFEST_BYTES` does not help here: it bounds a manifest parsed out of
/// an artifact, not one a caller constructed in Rust.
///
/// [`guest_text`](super::host::guest_text) is the bound this crate already
/// applies to text an untrusted party influenced, and it escapes as well as
/// truncates — so a name carrying U+202E cannot reorder the consent screen or
/// the log line that reports it refused.
pub(crate) fn rejected(value: &str) -> String {
    super::host::guest_text(value)
}

fn validate_prefix(prefix: &str) -> Result<(), ManifestError> {
    let refuse = |reason: &'static str| {
        Err(ManifestError::InvalidPrefix {
            prefix: rejected(prefix),
            reason,
        })
    };
    // Before the scan below, and before anything clones it: a prefix is
    // compared against every route path and then carried for the plugin's
    // lifetime, and a direct `from_module` caller chose its length.
    if prefix.len() > MAX_PATH_LEN {
        return refuse("a prefix must be at most 1024 bytes");
    }
    if !prefix.starts_with('/') {
        return refuse("a prefix must start with `/`");
    }
    if prefix == "/" {
        return refuse(
            "a plugin may not mount at the application root; give it a prefix of its own",
        );
    }
    if prefix.ends_with('/') {
        return refuse("a prefix must not end with `/`");
    }
    for segment in prefix.split('/').skip(1) {
        if segment.is_empty() {
            return refuse("a prefix must not contain an empty path segment");
        }
        if segment == "." || segment == ".." {
            return refuse("a prefix must not contain `.` or `..` segments");
        }
        if !segment
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.' | '~'))
        {
            return refuse(
                "a prefix must be a literal path: no wildcards, captures, query or fragment",
            );
        }
    }
    Ok(())
}

/// A segment that is a single whole capture (`{name}` / `{*name}`): the router
/// matches it by position, never by comparing its bytes to the request, so
/// byte-level rules for literal text do not apply to the name an author chose
/// for it. Anything else — including a *malformed* capture like `{id`, and
/// matchit's escaped braces (`{{café}}` is the literal text `{café}`) — is
/// literal text as far as the checks below are concerned, and the `matchit`
/// probe at the end still refuses a malformed one with the capture-specific
/// message.
fn is_capture_segment(segment: &str) -> bool {
    segment
        .strip_prefix('{')
        .and_then(|rest| rest.strip_suffix('}'))
        .is_some_and(|name| !name.contains(['{', '}']))
}

/// A declared route path may carry axum captures (`{id}`, `{*rest}`) — they are
/// matched by the host's router, never by the guest — but must otherwise be a
/// well-formed absolute path.
///
/// The capture syntax is checked by **inserting the path into a throwaway
/// `matchit` router**, which is the same engine axum 0.8 routes through, rather
/// than by a hand-written imitation of its rules. That matters more here than
/// anywhere else in this module: `axum::Router::route` *panics* on a template it
/// cannot insert, so a path this function waves through is a manifest that takes
/// the whole application down at boot. A plugin that can do that has defeated
/// the sandbox before it runs a single instruction.
fn validate_route_path(path: &str) -> Result<(), ManifestError> {
    let refuse = |reason: &'static str| {
        Err(ManifestError::InvalidRoutePath {
            path: rejected(path),
            reason,
        })
    };
    // Before the segment walk below and before the `matchit` insert that
    // follows it — both of which are linear in this path — and before the
    // accepted route is cloned into `OwnedRoutes`.
    if path.len() > MAX_PATH_LEN {
        return refuse("a route path must be at most 1024 bytes");
    }
    if !path.starts_with('/') {
        return refuse("a route path must start with `/`");
    }
    if path.len() > 1 && path.ends_with('/') {
        return refuse("a route path must not end with `/`");
    }
    if path.contains('?') || path.contains('#') {
        return refuse("a route path must not carry a query string or fragment");
    }
    for segment in path.split('/').skip(1) {
        if segment.is_empty() {
            return refuse("a route path must not contain an empty path segment");
        }
        // A route has one spelling, or the router's literal comparisons are
        // wrong. A client removes `%2e` as it removes `.`, and a proxy may
        // decode any ASCII escape or upper-case any hex, so the plugin could
        // mount beside the application route it arrives at (#2463).
        if dot_segment_len(segment).is_some() {
            return refuse("a route path must not contain `.` or `..` segments in any spelling");
        }
        if percent_escapes(segment).any(|(byte, _)| byte.is_ascii()) {
            return refuse("a route path must write an ASCII character as itself, not escaped");
        }
        if percent_escapes(segment).any(|(_, upper)| !upper) {
            return refuse("a route path must write a percent-escape in upper-case hex");
        }
        if segment.chars().any(char::is_whitespace) {
            return refuse("a route path must not contain whitespace");
        }
        // Not just whitespace: an ESC in a route path is printed verbatim on the
        // consent screen, where it can rewrite what the operator reads.
        if segment.chars().any(char::is_control) {
            return refuse("a route path must not contain control characters");
        }
        // `is_control` covers the C0/C1 categories and stops there, so it lets
        // through the Unicode formatting characters, which do the same job by
        // other means: U+202E reverses the run that follows it, so a path can
        // be made to *display* on the consent screen as something other than
        // what the manifest contains and the router mounts. The operator reads
        // that screen to decide whether to trust the artifact, so a character
        // whose only effect is on rendering is exactly the wrong thing to admit.
        if segment.chars().any(is_display_reordering) {
            return refuse(
                "a route path must not contain Unicode formatting characters; they change what \
                 the consent screen displays without changing what is mounted",
            );
        }
        // The router compares against the raw, percent-encoded path as it
        // arrived on the wire, while the manifest is written in decoded text.
        // A literal non-ASCII character can therefore never match: every
        // ordinary client encodes it before sending, so a manifest declaring
        // `/hello/café` mounts a route no client can reach while `plugin
        // inspect` prints it as a route the plugin serves (#2481). Refuse
        // rather than normalise, like every other rule in this function: the
        // encoded spelling is already the working one, and the consent screen
        // must show exactly what the router mounts. Captures (`{name}`,
        // `{*rest}`) are matched by position at request time, so the rule
        // applies to literal segments only. It runs after the whitespace,
        // control and formatting checks so those keep their more specific
        // refusals for the non-ASCII characters they cover.
        if !is_capture_segment(segment) && segment.bytes().any(|b| !b.is_ascii()) {
            return refuse(
                "a literal route segment must be ASCII; percent-encode each non-ASCII \
                 character as its upper-case UTF-8 bytes (e.g. `café` as `caf%C3%A9`)",
            );
        }
        // axum 0.8 spells captures `{name}` / `{*rest}` and *panics* on a
        // segment starting with the 0.7 spelling, before matchit ever sees it.
        // Naming the fix beats reporting matchit's message for a path it never
        // received.
        if segment.starts_with(':') {
            return refuse("axum 0.8 spells a capture `{name}`, not `:name`");
        }
        if segment.starts_with('*') {
            return refuse("axum 0.8 spells a catch-all `{*name}`, not `*name`");
        }
    }
    let mut probe: matchit::Router<()> = matchit::Router::new();
    if let Err(err) = probe.insert(path, ()) {
        return Err(ManifestError::InvalidRoutePath {
            path: rejected(path),
            reason: match err {
                matchit::InsertError::InvalidParam => {
                    "a capture must be spelled `{name}` with a non-empty name"
                }
                matchit::InsertError::InvalidCatchAll => {
                    "a catch-all `{*name}` is only allowed as the last segment"
                }
                matchit::InsertError::InvalidParamSegment => {
                    "a path segment may hold one whole capture and nothing else"
                }
                _ => "the router cannot mount this path",
            },
        });
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::plugin_sandbox::grants::{MAX_QUOTA, is_grantable_ident, is_grantable_name};

    fn valid_toml() -> String {
        format!(
            r#"
name = "autumn-plugin-hello"
version = "0.1.0"
wire_version = 1
prefix = "/hello"
capabilities = ["http-request"]
sha256 = "{digest}"

[[routes]]
method = "GET"
path = "/hello/greet"

[limits]
fuel = 200000000
memory_bytes = 33554432
max_request_body_bytes = 1048576
max_response_bytes = 4194304
max_concurrency = 8
"#,
            digest = "a".repeat(64)
        )
    }

    #[test]
    fn a_manifest_over_the_route_ceiling_is_refused_before_it_is_scanned() {
        // `MAX_MANIFEST_BYTES` bounds a manifest that was *parsed*, but
        // `SandboxHost::from_module` is public and takes a `SandboxManifest` an
        // embedder built in memory, where no byte ceiling applies. Validation
        // then scans the routes already seen for each new one, so the work
        // grows with the square of the list, and an accepted list is held again
        // for the lifetime of the plugin.
        let mut manifest = SandboxManifest::parse(&valid_toml()).expect("valid manifest");
        manifest.routes = (0..=MAX_ROUTES)
            .map(|index| DeclaredRoute {
                method: "GET".to_owned(),
                path: format!("/hello/r{index}"),
            })
            .collect();
        let err = manifest
            .validate()
            .expect_err("a manifest over the route ceiling must be refused");
        assert!(
            matches!(
                err,
                ManifestError::TooManyRoutes {
                    max: MAX_ROUTES,
                    ..
                }
            ),
            "{err}",
        );

        // Exactly at the ceiling still validates, so this bounds the list
        // rather than shrinking what a plugin may declare.
        let mut at_ceiling = SandboxManifest::parse(&valid_toml()).expect("valid manifest");
        at_ceiling.routes = (0..MAX_ROUTES)
            .map(|index| DeclaredRoute {
                method: "GET".to_owned(),
                path: format!("/hello/r{index}"),
            })
            .collect();
        at_ceiling
            .validate()
            .expect("a manifest at the ceiling must still validate");
    }

    #[test]
    fn a_route_path_past_the_byte_ceiling_is_refused_before_it_is_scanned() {
        // The count ceiling above bounds how many paths a manifest declares;
        // this bounds how large any one of them may be. On the count thread I
        // wrote that path length was "already bounded per route by
        // `validate_route_path`" — it was not, and the review that pushed back
        // on that was right: the function checked shape and never length, so a
        // direct `from_module` caller chose how much work the segment walk, the
        // `matchit` insert, and the `OwnedRoutes` clone each did.
        let long = format!("/hello/{}", "a".repeat(MAX_PATH_LEN));
        assert!(long.len() > MAX_PATH_LEN);
        let mut manifest = SandboxManifest::parse(&valid_toml()).expect("valid manifest");
        manifest.routes = vec![DeclaredRoute {
            method: "GET".to_owned(),
            path: long,
        }];
        let err = manifest
            .validate()
            .expect_err("a route path over the byte ceiling must be refused");
        assert!(
            matches!(err, ManifestError::InvalidRoutePath { .. }),
            "{err}",
        );

        // The refusal carries a bounded excerpt rather than the path it
        // objected to the size of — the whole point of `rejected`.
        assert!(
            err.to_string().len() < MAX_PATH_LEN,
            "the refusal is as large as the thing it refused: {} bytes",
            err.to_string().len(),
        );

        // Exactly at the ceiling still validates, so this bounds the path
        // rather than quietly shrinking what a plugin may declare.
        let mut at_ceiling = SandboxManifest::parse(&valid_toml()).expect("valid manifest");
        let exact = format!("/hello/{}", "a".repeat(MAX_PATH_LEN - "/hello/".len()));
        assert_eq!(exact.len(), MAX_PATH_LEN);
        at_ceiling.routes = vec![DeclaredRoute {
            method: "GET".to_owned(),
            path: exact,
        }];
        at_ceiling
            .validate()
            .expect("a route path at the ceiling must still validate");
    }

    #[test]
    fn a_prefix_past_the_byte_ceiling_is_refused_before_it_is_scanned() {
        // Same ceiling, same reason: the prefix is compared against every
        // declared route and then carried for the plugin's lifetime.
        let mut manifest = SandboxManifest::parse(&valid_toml()).expect("valid manifest");
        manifest.prefix = format!("/{}", "a".repeat(MAX_PATH_LEN));
        let err = manifest
            .validate()
            .expect_err("a prefix over the byte ceiling must be refused");
        assert!(matches!(err, ManifestError::InvalidPrefix { .. }), "{err}");
    }

    #[test]
    fn parses_a_valid_manifest() {
        let manifest = SandboxManifest::parse(&valid_toml()).expect("valid manifest");
        assert_eq!(manifest.name, "autumn-plugin-hello");
        assert_eq!(manifest.prefix, "/hello");
        assert_eq!(manifest.capabilities, vec![SandboxCapability::HttpRequest]);
        assert_eq!(manifest.routes.len(), 1);
        assert_eq!(manifest.limits.fuel, 200_000_000);
        assert!(manifest.is_granted(SandboxCapability::HttpRequest));
    }

    #[test]
    fn limits_default_when_the_section_is_absent() {
        let src = valid_toml();
        let trimmed = src.split("[limits]").next().expect("prefix").to_owned();
        let manifest = SandboxManifest::parse(&trimmed).expect("valid manifest");
        assert_eq!(manifest.limits, ResourceLimits::default());
    }

    #[test]
    fn an_unknown_capability_is_a_hard_error_naming_it() {
        let src = valid_toml().replace(r#"["http-request"]"#, r#"["http-request", "database"]"#);
        let err = SandboxManifest::parse(&src).expect_err("unknown capability must fail");
        let text = err.to_string();
        assert!(text.contains("database"), "{text}");
        assert!(text.contains("http-request"), "{text}");
    }

    #[test]
    fn an_unknown_manifest_key_is_a_hard_error() {
        let src = format!("{}\nallow_everything = true\n", valid_toml());
        let err = SandboxManifest::parse(&src).expect_err("unknown key must fail");
        assert!(err.to_string().contains("allow_everything"), "{err}");
    }

    #[test]
    fn a_manifest_without_the_http_request_capability_is_refused() {
        let src = valid_toml().replace(r#"["http-request"]"#, "[]");
        let err = SandboxManifest::parse(&src).expect_err("no capability must fail");
        assert!(matches!(err, ManifestError::MissingCapability(_)), "{err}");
    }

    #[test]
    fn a_future_wire_version_is_refused() {
        let src = valid_toml().replace("wire_version = 1", "wire_version = 2");
        let err = SandboxManifest::parse(&src).expect_err("wire version must fail");
        assert!(
            matches!(err, ManifestError::UnsupportedWireVersion { found: 2, .. }),
            "{err}"
        );
    }

    #[test]
    fn a_route_outside_the_declared_prefix_is_refused() {
        let src = valid_toml().replace(r#"path = "/hello/greet""#, r#"path = "/admin/users""#);
        let err = SandboxManifest::parse(&src).expect_err("off-prefix route must fail");
        assert!(
            matches!(err, ManifestError::RouteOutsidePrefix { .. }),
            "{err}"
        );
    }

    #[test]
    fn a_route_that_only_shares_a_prefix_string_is_refused() {
        let src = valid_toml().replace(r#"path = "/hello/greet""#, r#"path = "/helloworld""#);
        let err = SandboxManifest::parse(&src).expect_err("string-prefix route must fail");
        assert!(
            matches!(err, ManifestError::RouteOutsidePrefix { .. }),
            "{err}"
        );
    }

    #[test]
    fn the_prefix_itself_is_a_legal_route_path() {
        let src = valid_toml().replace(r#"path = "/hello/greet""#, r#"path = "/hello""#);
        assert!(SandboxManifest::parse(&src).is_ok());
    }

    #[test]
    fn a_root_prefix_is_refused() {
        let src = valid_toml()
            .replace(r#"prefix = "/hello""#, r#"prefix = "/""#)
            .replace(r#"path = "/hello/greet""#, r#"path = "/greet""#);
        let err = SandboxManifest::parse(&src).expect_err("root prefix must fail");
        assert!(matches!(err, ManifestError::InvalidPrefix { .. }), "{err}");
    }

    #[test]
    fn a_route_path_that_reorders_its_own_display_is_refused() {
        // U+202E makes the rendered path read differently from the mounted one.
        // The consent screen is where an operator decides whether to trust the
        // artifact, so a path that lies to it is refused outright.
        let toml =
            valid_toml().replace(r#"path = "/hello/greet""#, "path = \"/hello/\u{202E}terg\"");
        let err = SandboxManifest::parse(&toml).expect_err("must be refused");
        assert!(format!("{err}").contains("formatting characters"), "{err}");
    }

    #[test]
    fn a_prefix_with_a_wildcard_or_traversal_is_refused() {
        for bad in [
            "/he*llo",
            "/{tenant}",
            "/hello/",
            "/hello//x",
            "/../hello",
            "hello",
        ] {
            let src = valid_toml()
                .replace(r#"prefix = "/hello""#, &format!(r#"prefix = "{bad}""#))
                .replace(r#"path = "/hello/greet""#, &format!(r#"path = "{bad}""#));
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::InvalidPrefix { .. })
                ),
                "prefix {bad} must be refused"
            );
        }
    }

    #[test]
    fn a_route_path_the_router_would_refuse_is_refused_here() {
        // Each of these makes `axum::Router::route` panic. A manifest that
        // validates and then takes the app down at boot is the worst failure
        // this lane could have, so the validator has to speak the router's
        // language rather than a plausible imitation of it.
        for bad in [
            "/hello/:id",       // axum 0.7 capture syntax
            "/hello/*rest",     // axum 0.7 wildcard syntax
            "/hello/{id",       // unterminated capture
            "/hello/{}",        // unnamed capture
            "/hello/{*rest}/x", // catch-all that is not last
            "/hello/a{b}c",     // two things in one segment
        ] {
            let src =
                valid_toml().replace(r#"path = "/hello/greet""#, &format!(r#"path = "{bad}""#));
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::InvalidRoutePath { .. })
                ),
                "route path {bad} must be refused"
            );
        }
    }

    #[test]
    fn a_route_path_with_a_second_spelling_is_refused() {
        // #2463. Each of these names another route once a client or a proxy
        // normalises it. The router compares literals, so it would mount the
        // plugin beside the application route it arrives at.
        for bad in [
            "/hello/./transfer",
            "/hello/../transfer",
            "/hello/%2e/transfer",    // a single-dot segment, encoded
            "/hello/%2E/transfer",    // in upper case
            "/hello/%2e%2e/transfer", // a double-dot segment, encoded
            "/hello/.%2E/transfer",   // half encoded
            "/hello/%2e./transfer",   // the other half
            "/hello/a%2fb",           // an encoded `/`
            "/hello/a%2Fb",           // in upper case
            "/hello/a%5cb",           // an encoded `\`
            "/hello/%74ransfer",      // an encoded letter
            "/hello/%7E",             // an encoded unreserved symbol
            "/hello/a%20b",           // an encoded space
            "/hello/a%3Fb",           // an encoded `?`
            "/hello/%252e",           // an encoded `%`, one decode from `%2e`
            "/hello/caf%c3%a9",       // lower-case hex, which a proxy may upper-case
        ] {
            let src = valid_toml()
                .replace(r#"method = "GET""#, r#"method = "POST""#)
                .replace(r#"path = "/hello/greet""#, &format!(r#"path = "{bad}""#));
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::InvalidRoutePath { .. })
                ),
                "route path {bad} must be refused"
            );
        }

        // A non-ASCII character has no literal spelling in a URL, so its
        // escape is the one spelling it has.
        let src = valid_toml().replace(r#"path = "/hello/greet""#, r#"path = "/hello/caf%C3%A9""#);
        assert!(
            SandboxManifest::parse(&src).is_ok(),
            "an escaped `é` is its one spelling"
        );
    }

    #[test]
    fn a_capture_route_is_accepted() {
        let src = valid_toml().replace(r#"path = "/hello/greet""#, r#"path = "/hello/{name}""#);
        assert!(SandboxManifest::parse(&src).is_ok());
        let src = valid_toml().replace(r#"path = "/hello/greet""#, r#"path = "/hello/{*rest}""#);
        assert!(SandboxManifest::parse(&src).is_ok());
    }

    #[test]
    fn a_route_path_with_a_literal_non_ascii_segment_is_refused() {
        // #2481. The router compares against the raw, percent-encoded path as
        // it arrived on the wire, and every ordinary client encodes a
        // non-ASCII character before sending — so `/hello/café` mounts a
        // route no client can reach while `plugin inspect` prints it as a
        // route the plugin serves. The fix is refusal, not normalisation: the
        // encoded spelling is already the working one.
        // `{{…}}` is matchit's escaped-brace literal, not a capture, so its
        // text is compared against the wire like any other literal.
        for bad in [
            "/hello/café",
            "/café/x",
            "/hello/中文",
            "/😀",
            "/hello/{{café}}",
        ] {
            let src =
                valid_toml().replace(r#"path = "/hello/greet""#, &format!(r#"path = "{bad}""#));
            let err = SandboxManifest::parse(&src)
                .expect_err("a literal non-ASCII segment must be refused");
            assert!(
                matches!(err, ManifestError::InvalidRoutePath { .. }),
                "route path {bad} must be refused as an invalid route path: {err}"
            );
        }

        // The refusal says how to write the working spelling, so the author is
        // not left guessing. The message is input-independent (it cannot
        // name the rejected segment's own encoding), so it states the rule and
        // labels its `café` spelling as an example.
        let src = valid_toml().replace(r#"path = "/hello/greet""#, r#"path = "/hello/中文""#);
        let err = SandboxManifest::parse(&src).expect_err("must be refused");
        let message = format!("{err}");
        assert!(
            message.contains("percent-encode") && message.contains("e.g. `café` as `caf%C3%A9`"),
            "the refusal should explain the encoded spelling: {err}"
        );
    }

    #[test]
    fn an_encoded_non_ascii_segment_is_the_one_spelling_the_consent_screen_reports() {
        // The percent-encoded spelling is what clients send, so it validates —
        // and the consent screen must show exactly the spelling the router
        // mounts, or the screen lies about what the plugin serves (#2481).
        // Mounting takes `route.path` verbatim, so the assertion here is that
        // the consent text carries the same bytes the router gets.
        let src = valid_toml().replace(r#"path = "/hello/greet""#, r#"path = "/hello/caf%C3%A9""#);
        let manifest = SandboxManifest::parse(&src).expect("the encoded spelling is valid");
        assert!(
            manifest.consent_summary().contains("GET /hello/caf%C3%A9"),
            "the consent screen must report the mounted spelling"
        );
    }

    #[test]
    fn plain_ascii_routes_are_unaffected_by_the_non_ascii_rule() {
        // The new rule fires on non-ASCII bytes only: anything that validated
        // before still validates, punctuation and captures included.
        for good in [
            "/hello/greet",
            "/hello/a.b-c_d~e!f$g&h'i(j)k*l+m,n;o=p",
            "/hello/{name}",
            "/hello/{*rest}",
            "/hello/{id}/posts/{post_id}",
        ] {
            let src =
                valid_toml().replace(r#"path = "/hello/greet""#, &format!(r#"path = "{good}""#));
            assert!(
                SandboxManifest::parse(&src).is_ok(),
                "route path {good} must still validate"
            );
        }
    }

    #[test]
    fn two_routes_that_collide_in_the_router_are_refused() {
        // Distinct strings, one route as far as the router is concerned.
        for (first, second) in [("/hello/{a}", "/hello/{b}"), ("/hello/{*a}", "/hello/{b}")] {
            let src = format!(
                "{base}\n[[routes]]\nmethod = \"POST\"\npath = \"{second}\"\n",
                base = valid_toml()
                    .replace(r#"path = "/hello/greet""#, &format!(r#"path = "{first}""#)),
            );
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::ConflictingRoutes { .. })
                ),
                "{first} and {second} must be refused as a conflict"
            );
        }
    }

    #[test]
    fn the_same_path_under_two_methods_is_not_a_conflict() {
        let src = format!(
            "{base}\n[[routes]]\nmethod = \"POST\"\npath = \"/hello/{{name}}\"\n",
            base = valid_toml().replace(r#"path = "/hello/greet""#, r#"path = "/hello/{name}""#),
        );
        assert!(SandboxManifest::parse(&src).is_ok());
    }

    #[test]
    fn a_version_that_could_forge_a_consent_screen_is_refused() {
        // `version` is the one free-form field on the screen an operator reads
        // before agreeing to run the artifact.
        for bad in [
            "",
            "0.1.0 \u{1b}[2K",
            "0.1.0\u{7f}",
            "has space",
            &"9".repeat(200),
        ] {
            let src =
                valid_toml().replace(r#"version = "0.1.0""#, &format!(r#"version = "{bad}""#));
            assert!(
                SandboxManifest::parse(&src).is_err(),
                "version {bad:?} must be refused"
            );
        }
        assert!(SandboxManifest::parse(&valid_toml()).is_ok());
    }

    #[test]
    fn a_route_path_carrying_a_control_character_is_refused() {
        // A TOML `\u001B` escape decodes to a real ESC byte in the path — one
        // that would rewrite the consent screen an operator reads it from.
        let src = valid_toml().replace(
            r#"path = "/hello/greet""#,
            r#"path = "/hello/\u001B[2Kgreet""#,
        );
        assert!(
            matches!(
                SandboxManifest::parse(&src),
                Err(ManifestError::InvalidRoutePath { .. })
            ),
            "an escape sequence in a route path must be refused"
        );
    }

    #[test]
    fn an_invisible_letter_is_refused_though_no_format_class_contains_it() {
        // The reason the Cf-plus-exceptions shape kept leaking. U+115F HANGUL
        // CHOSEONG FILLER is general category Lo — a *letter*. It is not a
        // control, not whitespace, not a format character, and `is_alphabetic`
        // says yes about it. It also renders as nothing, which is the only
        // property that matters here.
        //
        // So the predicate is written against `Default_Ignorable_Code_Point`,
        // the property that actually names the hazard, rather than against a
        // general category that happens to contain most of it.
        for (ch, what) in [
            ('\u{115F}', "hangul choseong filler"),
            ('\u{1160}', "hangul jungseong filler"),
            ('\u{3164}', "hangul filler"),
            ('\u{FFA0}', "halfwidth hangul filler"),
        ] {
            assert!(
                ch.is_alphabetic(),
                "{what} is not a letter, so it does not test the gap that let it through",
            );
            assert!(
                is_display_reordering(ch),
                "{what} (U+{:04X}) renders as nothing and was admitted",
                ch as u32,
            );
        }
    }

    #[test]
    fn an_invisible_format_character_in_a_route_is_refused_like_a_visible_one() {
        // The predicate listed the bidi and zero-width ranges and stopped
        // there, which left the most literally invisible case through: U+00AD
        // SOFT HYPHEN renders as nothing, and neither `is_control` (Cc only)
        // nor `is_whitespace` claims it. `/hello/ad\u{00AD}min` therefore reads
        // as `/hello/admin` on the consent screen while mounting a path that is
        // not it — the same substitution the bidi overrides buy, with no
        // reordering needed.
        //
        // The whole format class is covered now, so this walks the classes
        // rather than the one character that was reported.
        for (ch, what) in [
            ('\u{00AD}', "soft hyphen"),
            ('\u{200B}', "zero-width space"),
            ('\u{200E}', "left-to-right mark"),
            ('\u{2062}', "invisible times"),
            ('\u{206F}', "nominal digit shapes"),
            ('\u{FEFF}', "zero-width no-break space"),
            ('\u{FFF9}', "interlinear annotation anchor"),
            ('\u{034F}', "combining grapheme joiner"),
            ('\u{FE0F}', "variation selector-16"),
            ('\u{E0041}', "tag latin capital A"),
            // Default-ignorable but *not* format characters, which is how the
            // Cf-plus-exceptions shape let them through a second time.
            ('\u{115F}', "hangul choseong filler"),
            ('\u{1160}', "hangul jungseong filler"),
            ('\u{3164}', "hangul filler"),
            ('\u{FFA0}', "halfwidth hangul filler"),
            ('\u{17B4}', "khmer vowel inherent aq"),
            ('\u{180F}', "mongolian free variation selector four"),
            ('\u{2065}', "reserved, default-ignorable"),
        ] {
            assert!(
                !ch.is_control() && !ch.is_whitespace(),
                "{what} is caught by an earlier check, so it does not test this one",
            );
            assert!(
                is_display_reordering(ch),
                "{what} (U+{:04X}) is display-altering and was admitted",
                ch as u32,
            );

            let src = valid_toml().replace(
                r#"path = "/hello/greet""#,
                &format!(r#"path = "/hello/ad{ch}min""#),
            );
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::InvalidRoutePath { .. })
                ),
                "a route path carrying {what} must be refused",
            );
        }

        // And ordinary text is unaffected — a path is allowed to be non-ASCII,
        // it is only allowed not to lie about what it is.
        for ch in ['é', 'ß', '日', '🦀'] {
            assert!(
                !is_display_reordering(ch),
                "{ch:?} is ordinary text and must not be refused",
            );
        }
    }

    #[test]
    fn a_manifest_with_no_routes_is_refused() {
        let src = valid_toml();
        let trimmed = src.split("[[routes]]").next().expect("prefix").to_owned();
        let err = SandboxManifest::parse(&trimmed).expect_err("no routes must fail");
        assert!(matches!(err, ManifestError::NoRoutes), "{err}");
    }

    #[test]
    fn an_unknown_http_method_is_refused() {
        let src = valid_toml().replace(r#"method = "GET""#, r#"method = "CONNECT""#);
        let err = SandboxManifest::parse(&src).expect_err("bad method must fail");
        assert!(matches!(err, ManifestError::InvalidMethod(_)), "{err}");
    }

    #[test]
    fn methods_are_normalised_to_upper_case() {
        let src = valid_toml().replace(r#"method = "GET""#, r#"method = "get""#);
        let manifest = SandboxManifest::parse(&src).expect("lowercase method is accepted");
        assert_eq!(manifest.routes[0].method, "GET");
    }

    #[test]
    fn duplicate_declared_routes_are_refused() {
        let src = format!(
            "{}\n[[routes]]\nmethod = \"GET\"\npath = \"/hello/greet\"\n",
            valid_toml()
        );
        let err = SandboxManifest::parse(&src).expect_err("duplicate route must fail");
        assert!(matches!(err, ManifestError::DuplicateRoute { .. }), "{err}");
    }

    #[test]
    fn a_capability_granted_twice_is_refused() {
        // A repeat conveys no authority the first grant did not, so there is no
        // manifest it makes legal — and the cost is not zero. Every request
        // hands the grant list to the frame and serialises it, work that
        // `request_footprint_bytes` never counted and `encoding_fuel` never
        // priced. Thousands of repeats fit inside the manifest size limit, and
        // a `SandboxManifest` built in-process has no size limit at all.
        let src = valid_toml().replace(
            r#"capabilities = ["http-request"]"#,
            r#"capabilities = ["http-request", "http-request"]"#,
        );
        let err = SandboxManifest::parse(&src).expect_err("a repeated capability must be refused");
        assert!(
            matches!(err, ManifestError::DuplicateCapability(_)),
            "{err}",
        );

        // Granted once, it still parses — the point is the repetition, not the
        // capability.
        SandboxManifest::parse(&valid_toml()).expect("one grant is the normal case");
    }

    #[test]
    fn a_malformed_digest_is_refused() {
        for bad in ["", "abc", &"A".repeat(64), &"z".repeat(64)] {
            let src = valid_toml().replace(&"a".repeat(64), bad);
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::InvalidDigest(_))
                ),
                "digest {bad} must be refused"
            );
        }
    }

    #[test]
    fn a_zero_or_oversized_limit_is_refused() {
        for (field, value) in [
            ("fuel", "0"),
            ("memory_bytes", "0"),
            ("max_concurrency", "0"),
            ("max_response_bytes", "0"),
            ("fuel", "999999999999999"),
            ("memory_bytes", "9999999999"),
        ] {
            let src = valid_toml()
                .lines()
                .map(|line| {
                    if line.starts_with(&format!("{field} = ")) {
                        format!("{field} = {value}")
                    } else {
                        line.to_owned()
                    }
                })
                .collect::<Vec<_>>()
                .join("\n");
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::LimitOutOfRange { .. })
                ),
                "{field} = {value} must be refused"
            );
        }
    }

    #[test]
    fn the_footprint_counts_the_host_buffers_a_request_holds_too() {
        // Linear memory is not the only thing a concurrent request pins: the
        // buffered request body, the pending stdout frame and the decoded
        // response all live in host memory outside the guest limiter. A
        // manifest with tiny `memory_bytes` and 64 MiB body/response ceilings
        // would otherwise pass the product check and still allocate hundreds of
        // gigabytes.
        let src = valid_toml()
            .replace("memory_bytes = 33554432", "memory_bytes = 65536")
            .replace(
                "max_request_body_bytes = 1048576",
                "max_request_body_bytes = 67108864",
            )
            .replace(
                "max_response_bytes = 4194304",
                "max_response_bytes = 67108864",
            )
            .replace("max_concurrency = 8", "max_concurrency = 1024");
        let err = SandboxManifest::parse(&src).expect_err("must be refused");
        assert!(
            matches!(err, ManifestError::LimitOutOfRange { field, .. } if field.contains("footprint")),
            "{err}"
        );
    }

    #[test]
    fn the_body_term_covers_the_base64_temporary_as_well_as_the_line() {
        // The term counted the caller's body, the frame's clone, and the
        // encoded text in the line — and missed that the encoding is built in a
        // `String` of its own first. `BASE64.encode(bytes)` allocates, and
        // `serialize_str` then copies that into the serializer's output, so
        // both are live at once and both are 4/3 of the body.
        //
        // Measured rather than asserted from arithmetic: the expansion has to
        // track what the encoder actually writes, so if that ever changes this
        // fails rather than quietly understating the product again.
        use crate::plugin_sandbox::wire::{HostFrame, SandboxRequest, to_line};

        let granted = [SandboxCapability::HttpRequest];
        let mut request = SandboxRequest {
            method: "GET".to_owned(),
            route: "/hello/greet".to_owned(),
            path: "/hello/greet".to_owned(),
            query: String::new(),
            path_params: vec![],
            headers: vec![("accept".to_owned(), "text/plain".to_owned())],
            body: vec![],
        };

        request.body = vec![b'x'; 30_000];
        let with = to_line(&HostFrame::request(&request, &granted))
            .expect("serialises")
            .len();
        request.body = Vec::new();
        let without = to_line(&HostFrame::request(&request, &granted))
            .expect("serialises")
            .len();

        // What one raw body byte becomes in the line: base64 is 4/3.
        let expansion = (with.saturating_sub(without) * 100) / 30_000;
        assert!(
            (130..=140).contains(&expansion),
            "the encoding is no longer ~4/3; the term below is derived from it: {expansion}%",
        );

        // Live at the peak: the caller's body, the frame's clone, the encoded
        // temporary, and the encoded text in the output — 1 + 1 + 4/3 + 4/3.
        let peak_percent = 200 + 2 * expansion;
        let limits = ResourceLimits {
            memory_bytes: 0,
            max_request_body_bytes: 1_000_000,
            max_response_bytes: 0,
            ..ResourceLimits::default()
        };
        let fixed = u128::from(crate::plugin_sandbox::host::MAX_TABLE_ELEMENTS) * 16
            + crate::plugin_sandbox::host::MAX_REQUEST_METADATA_BYTES as u128 * 8
            + crate::plugin_sandbox::host::MAX_GLOBALS as u128 * 16
            + crate::plugin_sandbox::host::MAX_FUNCTIONS as u128 * 32
            + crate::plugin_sandbox::host::FIXED_HOST_BUFFER_BYTES as u128;
        let charged_for_body = limits.request_footprint_bytes() - fixed;
        assert!(
            charged_for_body * 100 >= 1_000_000 * peak_percent as u128,
            "the body term ({charged_for_body}) is under the measured peak of \
             {peak_percent}% × the ceiling",
        );
    }

    #[test]
    fn the_footprint_counts_the_host_buffers_that_scale_with_nothing() {
        // Every other term scales with a ceiling the manifest names, so a
        // reviewer reading the manifest can see it. These do not scale with
        // anything, which is how they went missing: the stderr budget the state
        // holds for the whole request, and the scratch buffer an `fd_write`
        // allocates while that budget is still resident. A flat 4 KiB of slack
        // stood in for them and was two orders of magnitude short.
        //
        // Fixed per request is still per request. At a concurrency near this
        // product's own 1 GiB ceiling the difference is tens of megabytes of
        // host memory the advertised bound did not account for.
        use crate::plugin_sandbox::host::FIXED_HOST_BUFFER_BYTES;

        // A manifest that declares nothing: every scaling term is zero, so what
        // is left is exactly the fixed cost and nothing can hide inside it.
        let bare = ResourceLimits {
            memory_bytes: 0,
            max_request_body_bytes: 0,
            max_response_bytes: 0,
            ..ResourceLimits::default()
        };
        let scaling = u128::from(crate::plugin_sandbox::host::MAX_TABLE_ELEMENTS) * 16
            + crate::plugin_sandbox::host::MAX_REQUEST_METADATA_BYTES as u128 * 8
            + crate::plugin_sandbox::host::MAX_GLOBALS as u128 * 16
            + crate::plugin_sandbox::host::MAX_FUNCTIONS as u128 * 32;
        assert_eq!(
            bare.request_footprint_bytes() - scaling,
            FIXED_HOST_BUFFER_BYTES as u128,
        );

        // And the constant is really the buffers, not a number that drifted
        // away from them: it covers the stderr budget and one I/O chunk with
        // the denial ledger still to fit, and it is not back to the order of
        // the flat slack it replaced. Both are known at compile time, so they
        // are checked there — a term that stopped covering the buffers should
        // not build, rather than fail a test somebody has to run.
        const {
            assert!(
                FIXED_HOST_BUFFER_BYTES >= 64 * 1024 + 64 * 1024,
                "the fixed term no longer covers the stderr budget and one I/O chunk",
            );
            assert!(
                FIXED_HOST_BUFFER_BYTES > 4096 * 16,
                "the fixed term is back to standing in for the buffers rather than counting them",
            );
        }
    }

    #[test]
    fn the_footprint_counts_every_buffer_a_request_holds_at_once() {
        // The request body is buffered, cloned into the frame, base64-encoded
        // into a temporary, and copied from that into the NDJSON line that
        // becomes the guest's stdin — four live copies of an expanding thing,
        // not one — and the instance's tables are per-instance host storage
        // too.
        let limits = ResourceLimits {
            memory_bytes: 1_000_000,
            max_request_body_bytes: 100_000,
            max_response_bytes: 10_000,
            ..ResourceLimits::default()
        };
        let tables = u128::from(crate::plugin_sandbox::host::MAX_TABLE_ELEMENTS) * 16;
        let metadata = crate::plugin_sandbox::host::MAX_REQUEST_METADATA_BYTES as u128 * 8;
        let globals = crate::plugin_sandbox::host::MAX_GLOBALS as u128 * 16;
        let functions = crate::plugin_sandbox::host::MAX_FUNCTIONS as u128 * 32;
        let fixed = crate::plugin_sandbox::host::FIXED_HOST_BUFFER_BYTES as u128;
        assert_eq!(
            limits.request_footprint_bytes(),
            1_000_000 + 5 * 100_000 + 5 * 10_000 + tables + metadata + globals + functions + fixed
        );
    }

    #[test]
    fn the_footprint_counts_the_peak_while_a_response_is_being_decoded() {
        // Parsing the guest's answer is where the response side actually
        // peaks: the raw NDJSON line is still live (up to 2x the ceiling), the
        // base64 field may be copied out of it, and the decoded body is
        // allocated while both are held. A term that counted only "the line
        // plus the decoded response" described a moment that never happens.
        let limits = ResourceLimits {
            memory_bytes: 0,
            max_request_body_bytes: 0,
            max_response_bytes: 1_000_000,
            ..ResourceLimits::default()
        };
        let tables = u128::from(crate::plugin_sandbox::host::MAX_TABLE_ELEMENTS) * 16;
        let metadata = crate::plugin_sandbox::host::MAX_REQUEST_METADATA_BYTES as u128 * 8;
        let globals = crate::plugin_sandbox::host::MAX_GLOBALS as u128 * 16;
        let functions = crate::plugin_sandbox::host::MAX_FUNCTIONS as u128 * 32;
        let fixed = crate::plugin_sandbox::host::FIXED_HOST_BUFFER_BYTES as u128;
        assert_eq!(
            limits.request_footprint_bytes(),
            5 * 1_000_000 + tables + metadata + globals + functions + fixed,
            "the response term must cover the line, the base64 copy and the decode at once"
        );
    }

    #[test]
    fn the_footprint_counts_the_metadata_a_request_may_carry() {
        // The ceiling that bounds request metadata is the host's rather than
        // this manifest's, but it is per-request storage all the same, cloned
        // into the frame and serialised around. Left out, this product
        // understated a near-maximum-concurrency plugin by hundreds of
        // megabytes — and this product is exactly what the validator checks and
        // what a reviewer reads.
        let bare = ResourceLimits {
            memory_bytes: 0,
            max_request_body_bytes: 0,
            max_response_bytes: 0,
            ..ResourceLimits::default()
        };
        let metadata = crate::plugin_sandbox::host::MAX_REQUEST_METADATA_BYTES as u128 * 8;
        assert!(
            bare.request_footprint_bytes() >= metadata,
            "the metadata a request may carry is not in the footprint"
        );
    }

    #[test]
    fn the_default_limits_are_within_the_footprint_ceiling() {
        let manifest = SandboxManifest::parse(&valid_toml()).expect("valid");
        assert_eq!(manifest.limits, ResourceLimits::default());
    }

    #[test]
    fn a_value_rejected_for_its_size_is_not_copied_at_that_size() {
        // `SandboxManifest`'s fields are public, so a direct
        // `SandboxHost::from_module` caller reaches validation with a value the
        // container reader's `MAX_MANIFEST_BYTES` never saw. Rejecting a name
        // for being too long, and then carrying the whole thing into the error
        // to say so, holds the original plus a copy plus the string
        // `from_module` formats out of it — three of the thing whose size was
        // the objection.
        //
        // The bound is what is asserted, not the exact excerpt: the point is
        // that the error's size is decided by this crate rather than by the
        // caller.
        let huge = "x".repeat(512 * 1024);

        let mut manifest = SandboxManifest::parse(&valid_toml()).expect("valid");
        manifest.name.clone_from(&huge);
        let err = manifest
            .validate()
            .expect_err("an oversized name is refused");
        let rendered = err.to_string();
        assert!(
            rendered.len() < huge.len() / 100,
            "the error carries the rejected value at its own size: {} bytes for a {} byte name",
            rendered.len(),
            huge.len(),
        );

        // The sibling fields reject on shape too, and each was its own copy.
        let mut manifest = SandboxManifest::parse(&valid_toml()).expect("valid");
        manifest.sha256.clone_from(&huge);
        let rendered = manifest
            .validate()
            .expect_err("an oversized digest is refused")
            .to_string();
        assert!(
            rendered.len() < huge.len() / 100,
            "the digest error carries the rejected value: {} bytes",
            rendered.len(),
        );

        let mut manifest = SandboxManifest::parse(&valid_toml()).expect("valid");
        manifest.prefix = format!("/{huge}");
        let rendered = manifest
            .validate()
            .expect_err("an oversized prefix is refused")
            .to_string();
        assert!(
            rendered.len() < huge.len() / 100,
            "the prefix error carries the rejected value: {} bytes",
            rendered.len(),
        );
    }

    #[test]
    fn a_name_that_could_forge_a_log_line_is_refused() {
        for bad in ["", "a b", "../etc", "plugin:name", &"x".repeat(200)] {
            let src = valid_toml().replace("autumn-plugin-hello", bad);
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::InvalidName(_))
                ),
                "name {bad:?} must be refused"
            );
        }
    }

    #[test]
    fn a_name_carrying_a_newline_is_refused() {
        // TOML itself refuses a raw newline inside a basic string, so this one
        // never reaches the name check — but it must still be a refusal, and
        // the test exists so a future manifest format that *does* accept it
        // cannot silently let a log-forging name through.
        let src = valid_toml().replace("autumn-plugin-hello", "plugin\nname");
        assert!(SandboxManifest::parse(&src).is_err());
    }

    #[test]
    fn round_trips_through_toml() {
        let manifest = SandboxManifest::parse(&valid_toml()).expect("valid");
        let rendered = manifest.to_toml().expect("serializes");
        let reparsed = SandboxManifest::parse(&rendered).expect("re-parses");
        assert_eq!(manifest, reparsed);
    }

    // ── Capability vocabulary (issue #1632) ──────────────────────────

    /// The manifest for a plugin that asks for the whole grown vocabulary.
    fn vocabulary_toml() -> String {
        format!(
            r#"
name = "autumn-plugin-shop"
version = "0.1.0"
wire_version = 1
prefix = "/shop"
capabilities = ["http-request", "kv", "http-outbound", "db", "jobs", "render"]
sha256 = "{digest}"

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

[grants]
hosts = ["api.example.com"]
tables = ["orders"]
job_types = ["reindex"]
slots = ["order-summary"]
"#,
            digest = "b".repeat(64)
        )
    }

    #[test]
    fn the_vocabulary_covers_kv_outbound_db_jobs_and_render() {
        let manifest = SandboxManifest::parse(&vocabulary_toml()).expect("valid");
        for capability in [
            SandboxCapability::HttpRequest,
            SandboxCapability::Kv,
            SandboxCapability::HttpOutbound,
            SandboxCapability::Db,
            SandboxCapability::Jobs,
            SandboxCapability::Render,
        ] {
            assert!(manifest.is_granted(capability), "{capability}");
            assert!(SandboxCapability::ALL.contains(&capability), "{capability}");
        }
    }

    #[test]
    fn every_capability_round_trips_through_its_manifest_spelling() {
        for capability in SandboxCapability::ALL {
            assert_eq!(
                SandboxCapability::parse(capability.as_str()),
                Ok(*capability)
            );
            assert!(!capability.describe().is_empty());
        }
    }

    #[test]
    fn a_grant_list_without_its_capability_is_refused() {
        // The operator read "no outbound network" in the capability list and
        // "api.example.com" three lines below it. One of those is a lie, and
        // the runtime must not pick which.
        let src = vocabulary_toml().replace(r#", "http-outbound""#, "");
        assert_eq!(
            SandboxManifest::parse(&src),
            Err(ManifestError::GrantWithoutCapability {
                capability: SandboxCapability::HttpOutbound,
                field: "hosts",
            })
        );
    }

    #[test]
    fn a_capability_with_an_empty_grant_list_is_refused() {
        let src = vocabulary_toml().replace(r#"hosts = ["api.example.com"]"#, "hosts = []");
        assert_eq!(
            SandboxManifest::parse(&src),
            Err(ManifestError::CapabilityWithoutGrant {
                capability: SandboxCapability::HttpOutbound,
                field: "hosts",
            })
        );
    }

    #[test]
    fn an_outbound_host_carrying_a_scheme_port_or_path_is_refused() {
        for bad in [
            "https://api.example.com",
            "api.example.com:443",
            "api.example.com/v1",
            "user@api.example.com",
            "API.example.com",
            "",
            "-api.example.com",
            "api..example.com",
            "*.example.com",
        ] {
            let src = vocabulary_toml().replace("api.example.com", bad);
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::InvalidGrantEntry { .. })
                ),
                "{bad} was accepted"
            );
        }
    }

    #[test]
    fn a_table_job_type_or_slot_that_is_not_a_plain_identifier_is_refused() {
        for (field, bad) in [
            ("orders", "orders; drop table users"),
            ("orders", "orders--"),
            ("orders", "Orders"),
            ("orders", "public.users"),
            ("reindex", "re index"),
            ("order-summary", "order summary"),
            ("order-summary", "Order-Summary"),
        ] {
            let src = vocabulary_toml().replace(field, bad);
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::InvalidGrantEntry { .. })
                ),
                "{bad} was accepted for {field}"
            );
        }
        // A name carrying a quote cannot even be written in the manifest's own
        // syntax without escaping, so the shape rule is asserted directly as
        // well — `SandboxHost::from_module` takes a manifest an embedder built
        // in memory, where TOML never ran.
        for bad in ["\"orders\"", "orders`", "orders'", "ord ers", ""] {
            assert!(!is_grantable_ident(bad), "{bad} is not an identifier");
            assert!(!is_grantable_name(bad), "{bad} is not a name");
        }
    }

    #[test]
    fn a_repeated_grant_entry_is_refused() {
        let src =
            vocabulary_toml().replace(r#"tables = ["orders"]"#, r#"tables = ["orders", "orders"]"#);
        assert!(matches!(
            SandboxManifest::parse(&src),
            Err(ManifestError::DuplicateGrantEntry { .. })
        ));
    }

    #[test]
    fn quotas_default_conservatively_and_are_operator_configurable() {
        let manifest = SandboxManifest::parse(&vocabulary_toml()).expect("valid");
        let defaults = CapabilityQuotas::default();
        assert_eq!(manifest.quotas, defaults);
        assert!(defaults.kv_reads > 0 && defaults.kv_reads <= MAX_QUOTA);

        let src = format!("{}\n[quotas]\nkv_reads = 3\n", vocabulary_toml());
        let raised = SandboxManifest::parse(&src).expect("valid");
        assert_eq!(raised.quotas.kv_reads, 3);
        // Everything the operator did not name keeps its conservative default.
        assert_eq!(raised.quotas.kv_writes, defaults.kv_writes);
    }

    #[test]
    fn a_zero_or_oversized_quota_is_refused() {
        for value in ["0", "4294967295"] {
            let src = format!("{}\n[quotas]\nkv_reads = {value}\n", vocabulary_toml());
            assert!(
                matches!(
                    SandboxManifest::parse(&src),
                    Err(ManifestError::QuotaOutOfRange { .. })
                ),
                "{value} was accepted"
            );
        }
    }

    #[test]
    fn an_upgrade_that_asks_for_more_needs_fresh_consent() {
        // Same plugin, same routes, same ceilings — only the grant grows, so
        // the delta isolates the thing under test. Diffing two *different*
        // plugins would report every route as new and prove nothing about
        // capabilities.
        let previous = SandboxManifest::parse(
            &vocabulary_toml()
                .replace(
                    r#"capabilities = ["http-request", "kv", "http-outbound", "db", "jobs", "render"]"#,
                    r#"capabilities = ["http-request"]"#,
                )
                .replace(r#"hosts = ["api.example.com"]"#, "")
                .replace(r#"tables = ["orders"]"#, "")
                .replace(r#"job_types = ["reindex"]"#, "")
                .replace(r#"slots = ["order-summary"]"#, ""),
        )
        .expect("valid");
        let next = SandboxManifest::parse(&vocabulary_toml()).expect("valid");

        let delta = next.consent_delta_from(&previous);
        assert!(delta.requires_consent());
        assert!(delta.added_capabilities.contains(&SandboxCapability::Kv));
        assert!(
            delta
                .added_hosts
                .iter()
                .any(|host| host == "api.example.com")
        );
        assert!(delta.added_tables.iter().any(|table| table == "orders"));
        assert!(delta.added_job_types.iter().any(|job| job == "reindex"));
        assert!(delta.added_slots.iter().any(|slot| slot == "order-summary"));
        assert!(
            delta.added_routes.is_empty() && delta.raised_limits.is_empty(),
            "only the grant moved: {delta:?}"
        );
        let summary = delta.summary();
        assert!(summary.contains("api.example.com"), "{summary}");

        // The same manifest twice asks for nothing new.
        assert!(!next.consent_delta_from(&next).requires_consent());
        // Dropping a capability is not something to re-prompt for.
        assert!(!previous.consent_delta_from(&next).requires_consent());
    }

    #[test]
    fn raising_a_quota_needs_fresh_consent_but_lowering_one_does_not() {
        let base = format!("{}\n[quotas]\nkv_writes = 4\n", vocabulary_toml());
        let raised = format!("{}\n[quotas]\nkv_writes = 40\n", vocabulary_toml());
        let base = SandboxManifest::parse(&base).expect("valid");
        let raised = SandboxManifest::parse(&raised).expect("valid");

        let up = raised.consent_delta_from(&base);
        assert!(up.requires_consent());
        assert!(
            up.raised_quotas
                .iter()
                .any(|(field, ..)| *field == "kv_writes")
        );
        assert!(!base.consent_delta_from(&raised).requires_consent());
    }

    #[test]
    fn an_upgrade_that_mounts_a_new_route_needs_fresh_consent() {
        // A route is the router, not a description of it: the host builds its
        // mount from exactly this list, and the consent screen promises the
        // plugin serves "these and only these". An upgrade adding one exposes
        // an endpoint nobody approved.
        let previous = SandboxManifest::parse(&vocabulary_toml()).expect("valid");
        let next = SandboxManifest::parse(&format!(
            "{}\n[[routes]]\nmethod = \"POST\"\npath = \"/shop/checkout\"\n",
            vocabulary_toml()
        ))
        .expect("valid");

        let delta = next.consent_delta_from(&previous);
        assert!(delta.requires_consent());
        assert!(
            delta
                .added_routes
                .iter()
                .any(|route| route == "POST /shop/checkout"),
            "{:?}",
            delta.added_routes
        );
        assert!(delta.summary().contains("/shop/checkout"));
        // Dropping a route asks for less.
        assert!(!previous.consent_delta_from(&next).requires_consent());
    }

    #[test]
    fn the_head_a_new_get_implies_is_part_of_the_upgrade_too() {
        // HTTP serves HEAD wherever it serves GET and the runtime mounts it, so
        // a manifest adding one bare GET adds two mounted routes. Both are
        // authority, and the diff is taken over the list the consent screen
        // prints rather than the literal `[[routes]]`.
        let previous = SandboxManifest::parse(&vocabulary_toml()).expect("valid");
        let next = SandboxManifest::parse(&format!(
            "{}\n[[routes]]\nmethod = \"GET\"\npath = \"/shop/orders\"\n",
            vocabulary_toml()
        ))
        .expect("valid");
        let delta = next.consent_delta_from(&previous);
        assert!(
            delta.added_routes.contains(&"GET /shop/orders".to_owned())
                && delta.added_routes.contains(&"HEAD /shop/orders".to_owned()),
            "{:?}",
            delta.added_routes
        );
    }

    #[test]
    fn a_quota_for_a_dropped_capability_is_not_new_authority() {
        // The narrowing upgrade this delta exists to wave through, made to look
        // like growth by a number left behind. Dropping `kv` while a raised
        // `kv_reads` stays in the table is not new authority — no KV call can
        // be made at all — but it made `plugin inspect --against` exit
        // non-zero, prompting on the change least in need of a prompt.
        let approved =
            SandboxManifest::parse(&format!("{}\n[quotas]\nkv_reads = 8\n", vocabulary_toml()))
                .expect("valid");

        // Same manifest with `kv` and its grants gone, and `kv_reads` raised.
        let narrowed_src = vocabulary_toml().replace(
            r#"capabilities = ["http-request", "kv", "http-outbound", "db", "jobs", "render"]"#,
            r#"capabilities = ["http-request", "http-outbound", "db", "jobs", "render"]"#,
        );
        let narrowed =
            SandboxManifest::parse(&format!("{narrowed_src}\n[quotas]\nkv_reads = 64\n"))
                .expect("valid");

        let delta = narrowed.consent_delta_from(&approved);
        assert!(
            !delta.requires_consent(),
            "dropping a capability is not growth: {delta:?}"
        );
        assert!(
            !delta
                .raised_quotas
                .iter()
                .any(|(field, ..)| *field == "kv_reads"),
            "{:?}",
            delta.raised_quotas
        );

        // And the same raise *with* `kv` still granted is still reported, so
        // the filter narrows the delta rather than defeating it.
        let greedy =
            SandboxManifest::parse(&format!("{}\n[quotas]\nkv_reads = 64\n", vocabulary_toml()))
                .expect("valid");
        assert!(
            greedy
                .consent_delta_from(&approved)
                .raised_quotas
                .iter()
                .any(|(field, ..)| *field == "kv_reads"),
            "a raise on a capability that is still granted must still prompt"
        );
    }

    #[test]
    fn every_quota_says_which_capability_it_bounds() {
        // The pairing `governed_by` relies on, asserted rather than assumed:
        // a quota added to `fields` without a mapping would silently become
        // ungoverned, and an ungoverned quota is always reported — which is the
        // behaviour the filter above exists to remove.
        let ungoverned = ["calls", "calls_per_second"];
        for (field, _) in CapabilityQuotas::default().fields() {
            let governed = super::super::grants::CapabilityQuotas::governed_by(field).is_some();
            assert_eq!(
                governed,
                !ungoverned.contains(&field),
                "{field} is on the wrong side of the governed/ungoverned split"
            );
        }
    }

    #[test]
    fn an_upgrade_that_raises_a_resource_ceiling_needs_fresh_consent() {
        // The kind of growth that touches no capability name: `fuel`,
        // `memory_bytes` and `max_concurrency` are what one plugin may cost the
        // host, and an upgrade can multiply them by thousands while every word
        // on the consent screen stays the same.
        let modest = SandboxManifest::parse(&format!(
            "{}\n[limits]\nfuel = 5000000\nmemory_bytes = 16777216\nmax_concurrency = 4\n",
            vocabulary_toml()
        ))
        .expect("valid");
        let greedy = SandboxManifest::parse(&format!(
            "{}\n[limits]\nfuel = 100000000000\nmemory_bytes = 16777216\nmax_concurrency = 4\n",
            vocabulary_toml()
        ))
        .expect("valid");

        let delta = greedy.consent_delta_from(&modest);
        assert!(delta.requires_consent());
        assert!(
            delta
                .raised_limits
                .iter()
                .any(|(field, ..)| *field == "fuel"),
            "{:?}",
            delta.raised_limits
        );
        assert!(delta.summary().contains("fuel"));
        // Lowering a ceiling asks for less.
        assert!(!modest.consent_delta_from(&greedy).requires_consent());
    }

    #[test]
    fn a_db_grant_whose_physical_name_cannot_be_derived_is_refused_at_load() {
        // The alternative is a capability the consent screen displays and every
        // call then denies as `malformed` — an operator approving authority the
        // runtime can never honour.
        let src = vocabulary_toml()
            .replace("autumn-plugin-shop", &"p".repeat(40))
            .replace("\"orders\"", &format!("\"{}\"", "t".repeat(30)));
        assert!(
            matches!(
                SandboxManifest::parse(&src),
                Err(ManifestError::InvalidGrantEntry {
                    field: "tables",
                    ..
                })
            ),
            "{:?}",
            SandboxManifest::parse(&src)
        );
    }

    #[test]
    fn the_consent_summary_enumerates_every_grant_detail_and_quota() {
        let manifest = SandboxManifest::parse(&vocabulary_toml()).expect("valid");
        let summary = manifest.consent_summary();
        for expected in [
            "api.example.com",
            "orders",
            "reindex",
            "order-summary",
            "kv_reads",
        ] {
            assert!(
                summary.contains(expected),
                "{expected} missing from {summary}"
            );
        }
        // The blanket "no database, no network" line must not survive next to a
        // manifest that was just granted both.
        assert!(
            !summary.contains("filesystem access, outbound network access"),
            "{summary}"
        );
    }

    #[test]
    fn a_plugin_that_asks_for_nothing_extra_still_reads_as_denying_everything() {
        let summary = SandboxManifest::parse(&valid_toml())
            .expect("valid")
            .consent_summary();
        for denied in ["filesystem", "network", "environment", "database"] {
            assert!(summary.contains(denied), "{denied} missing from {summary}");
        }
    }

    #[test]
    fn the_consent_summary_names_the_grant_the_prefix_and_the_digest() {
        let manifest = SandboxManifest::parse(&valid_toml()).expect("valid");
        let summary = manifest.consent_summary();
        assert!(summary.contains("autumn-plugin-hello"), "{summary}");
        assert!(summary.contains("/hello"), "{summary}");
        assert!(summary.contains("http-request"), "{summary}");
        assert!(summary.contains(&"a".repeat(64)), "{summary}");
        assert!(summary.contains("GET /hello/greet"), "{summary}");
        // Everything the sandbox denies is named too, so the reader sees the
        // shape of the "no" and not just the "yes".
        assert!(summary.contains("filesystem"), "{summary}");
        assert!(summary.contains("network"), "{summary}");
        assert!(summary.contains("environment"), "{summary}");
        assert!(summary.contains("database"), "{summary}");
    }

    #[test]
    fn a_declared_get_reports_the_head_it_also_serves() {
        // HTTP says HEAD is GET without the body, and axum's method router
        // dispatches a HEAD with no HEAD route to the GET one. A manifest that
        // listed only GET would therefore serve a method its own consent screen
        // never named.
        let manifest = SandboxManifest::parse(&valid_toml()).expect("valid");
        let infos = manifest.route_infos();
        assert_eq!(infos.len(), 2, "{infos:?}");
        assert!(infos.iter().any(|route| route.method == "HEAD"));
        assert!(manifest.consent_summary().contains("HEAD /hello/greet"));
    }

    #[test]
    fn route_infos_carry_plugin_attribution_under_the_prefix() {
        let manifest = SandboxManifest::parse(&valid_toml()).expect("valid");
        let infos = manifest.route_infos();
        assert_eq!(infos.len(), 2);
        assert_eq!(infos[0].method, "GET");
        assert_eq!(infos[0].path, "/hello/greet");
        assert_eq!(
            infos[0].source,
            crate::route_listing::RouteSource::Plugin("autumn-plugin-hello".to_owned())
        );
    }
}