zenkey-fleet 0.16.0

//! The ACL planner (RFC 09 §3, #392): an enrollment file plus, when given,
//! the registry in; the router's `access_control` block out — with every
//! rule naming the matrix row it instantiates and the fact it exists for,
//! every narrowing shown, and every refusal named.
//!
//! RFC 09 §3 has carried a complete recipe since v1.0 and it went undeployed
//! for a year, because writing it by hand is miserable and unverifiable.
//! Four facts shape it, and each one is a way a hand-written block fails
//! **silently and partially**:
//!
//! 1. Matching is keyexpr *inclusion*, and `**` never crosses a verbatim
//!    chunk there either — a host's `…/h-xxx/**` rule does not cover its
//!    `@rpc` replies, `@media` frames or `@blob` keys. One rule per plane.
//! 2. `*` in the origin position never covers `@catalog`; catalog access is
//!    its own rule (RFC 03 §4 D4).
//! 3. The config needs all three lists — `rules`, `subjects`, `policies`;
//!    rules alone are refused at router startup.
//! 4. Under `default_permission: "deny"`, *declarations* need allowing: a
//!    consumer that may not `declare_subscriber` receives nothing, a
//!    producer that may not `declare_queryable` serves nothing, and a query
//!    must be allowed *egress* toward the responder as well as ingress from
//!    the caller.
//!
//! And a fifth, from the reference deployment (2026-08-30), RFC 09 §3
//! fact 5 since v1.33: zenoh evaluates a consumer's declares and queries **on egress
//! toward the responding face, against that face's subject**. A host whose
//! grants are all own-origin (`…/h-xxx/**`) includes no wildcard-origin
//! selector, so a console's `…/v1/**` interest never reaches it, and a
//! peer-mode publisher with no matching interest publishes to nobody. The
//! fix is one shared, **egress-only** rule — [`INTEREST_PROP`] — allowing
//! declares and queries on the fleet's selectors, attached to every
//! publishing principal. It is security-neutral: puts and replies stay
//! ingress-checked. The same fact has a corollary: `**` cannot cross
//! `@catalog` any more than `@adv`, so `@catalog/**/@adv/**` needs spelling
//! out wherever the catalog runs the advanced tier.
//!
//! And a sixth (RFC 09 §3 fact 6, v1.38): **deny is by inclusion**, so
//! [`NO_REMOTE_ACTIONS`] binds only a caller who spells a write's key at
//! least as narrowly as the rule. A query on `…/v1/*/@rpc/**` — which the
//! console's own fleet grant includes — is not included by the literal, and
//! crosses to every queryable it intersects. The rule stays, as the second
//! lock on the door; the refusal of a broadcast write is the *server's*
//! (RFC 05 §2.1, `BringUp::serve_write`), and the rule's cite says so.
//!
//! Pure, like everything in [`crate::model`]: values in hand, no session.
//! [`plan_acl`] takes an *optional* registry and says what it could not
//! narrow without one rather than guessing (RFC 13 §3 O4); [`check_acl`]
//! compares a plan against a block somebody else read off a router's config
//! file; [`explain_acl`] answers "does this principal hold this grant, via
//! which rule, in which direction" over the plan alone, with inclusion
//! computed by `zenoh-keyexpr`. [`to_json5`] is the one rendering of the
//! plan that is not zenkey's — it is `zenohd`'s, field for field
//! `zenoh-config-1.10.0/src/lib.rs`.

use std::collections::{BTreeMap, BTreeSet};

use zenoh::key_expr::keyexpr;

use crate::model::registry::SliceSet;
use crate::report::{
    AclCheck, AclConfigDoc, AclDecision, AclDirection, AclDownsample, AclExplain, AclFinding,
    AclFindingKind, AclFlow, AclGrant, AclMessage, AclPermission, AclPlan, AclPolicy, AclRefusal,
    AclRegistryFacts, AclRule, AclSubject, AclWarning, AclWarningKind, Asked, Enrollment, FaceSpec,
    Judgement, LinkInterval, PrincipalSpec, Role,
};
use crate::{Error, Result};

/// The shared egress-only rule every publishing principal carries — the
/// fifth fact (module doc).
pub const INTEREST_PROP: &str = "interest-prop";
/// The shared deny on the write procedures — the console's
/// `no-remote-actions` row, narrowed by the registry when one was asked.
pub const NO_REMOTE_ACTIONS: &str = "no-remote-actions";
/// The convention's write-procedure leaf (RFC 09 §3, v1.4: a write is keyed
/// `…/set` so a rule can see the actuated resource) — the deny's shape when
/// no registry narrowed it.
pub const UNNARROWED_WRITE_LEAF: &str = "v1/*/@rpc/*/**/set";

/// What the caller decided about the enrollment's dangerous corners.
#[derive(Debug, Clone, Copy, Default)]
pub struct AclOptions {
    /// Admit `zid = "…"` subjects. Off by default: a ZID is not backed by
    /// authentication (zenoh's own config says so).
    pub allow_zid_subjects: bool,
}

// ── The matrix rows ───────────────────────────────────────────────────────

const DECLARES: [AclMessage; 4] = [
    AclMessage::DeclareSubscriber,
    AclMessage::DeclareLivelinessSubscriber,
    AclMessage::LivelinessQuery,
    AclMessage::Query,
];
const RECEIVES: [AclMessage; 4] = [
    AclMessage::Put,
    AclMessage::Delete,
    AclMessage::Reply,
    AclMessage::LivelinessToken,
];

fn wire(base: &str, rel: impl AsRef<str>) -> String {
    zenkey::grammar::with_base(base, rel)
}

/// The planes a consumer names, each spelled because `**` reaches none of
/// them from `v1/**` (fact 1) and `*` never reaches `@catalog` (fact 2).
struct Planes {
    media: bool,
    blob: bool,
    catalog_adv: bool,
}

fn fleet_exprs(base: &str, p: &Planes) -> Vec<String> {
    let mut out = vec![
        wire(base, "v1/**"),
        wire(base, "v1/@catalog/**"),
        wire(base, "v1/*/@rpc/**"),
        wire(base, "v1/@catalog/@rpc/**"),
    ];
    if p.blob {
        out.push(wire(base, "v1/*/@blob/**"));
    }
    if p.media {
        out.push(wire(base, "v1/*/@media/**"));
    }
    out.push(wire(base, "v1/**/@adv/**"));
    if p.catalog_adv {
        out.push(wire(base, "v1/@catalog/**/@adv/**"));
    }
    out
}

/// The `@adv` sidecar expressions, catalog sibling included when the
/// catalog runs the tier.
fn adv_exprs(base: &str, catalog_adv: bool) -> Vec<String> {
    let mut out = vec![wire(base, "v1/**/@adv/**")];
    if catalog_adv {
        out.push(wire(base, "v1/@catalog/**/@adv/**"));
    }
    out
}

fn rule(
    id: impl Into<String>,
    permission: AclPermission,
    flows: Option<&[AclFlow]>,
    messages: &[AclMessage],
    key_exprs: Vec<String>,
    purpose: &str,
    cite: &str,
) -> AclRule {
    AclRule {
        id: id.into(),
        permission,
        flows: flows.map(<[AclFlow]>::to_vec),
        messages: messages.to_vec(),
        key_exprs,
        purpose: purpose.into(),
        cite: cite.into(),
    }
}

const IN: &[AclFlow] = &[AclFlow::Ingress];
const OUT: &[AclFlow] = &[AclFlow::Egress];

/// A service origin's chunk without its `@`, for rule ids.
fn id_chunk(origin: &str) -> &str {
    origin.strip_prefix('@').unwrap_or(origin)
}

// ── The registry's part ───────────────────────────────────────────────────

fn registry_facts(slices: &SliceSet) -> AclRegistryFacts {
    let mut media_producers = Vec::new();
    let mut blob_producers = Vec::new();
    let mut write_procedures = Vec::new();
    let mut sensitive_procedures = Vec::new();
    for slice in slices.slices() {
        if slice.service_origin.is_none() {
            if !slice.media.is_empty() {
                media_producers.push(slice.name.clone());
            }
            if !slice.blob.is_empty() {
                blob_producers.push(slice.name.clone());
            }
        }
        for p in &slice.procedures {
            if p.kind
                .as_ref()
                .is_some_and(|k| k.is(&zenkey::slice::ProcedureKind::Write))
            {
                write_procedures.push(format!("{}/{}", slice.name, p.path));
                if p.sensitive == Some(true) {
                    sensitive_procedures.push(format!("{}/{}", slice.name, p.path));
                }
            }
        }
    }
    media_producers.sort();
    blob_producers.sort();
    write_procedures.sort();
    sensitive_procedures.sort();
    AclRegistryFacts {
        slices: slices.slices().len(),
        media_producers,
        blob_producers,
        write_procedures,
        sensitive_procedures,
    }
}

/// A registry path as a key expression: every `{var}` chunk is a `*`, and a
/// `{var...}` rest chunk (subjects only; refused on procedures) is a `**`.
fn procedure_pattern(path: &str) -> String {
    path.split('/')
        .map(|c| {
            if c.starts_with('{') && c.ends_with("...}") {
                "**"
            } else if c.starts_with('{') && c.ends_with('}') {
                "*"
            } else {
                c
            }
        })
        .collect::<Vec<_>>()
        .join("/")
}

/// Every declared write procedure's key expression, under the base.
fn write_exprs(base: &str, slices: &SliceSet) -> Vec<String> {
    procedure_exprs(base, slices, |p| {
        p.kind
            .as_ref()
            .is_some_and(|k| k.is(&zenkey::slice::ProcedureKind::Write))
    })
}

/// The `sensitive = true` writes (RFC 08 §2, v1.43), under the base.
fn sensitive_exprs(base: &str, slices: &SliceSet) -> Vec<String> {
    procedure_exprs(base, slices, |p| {
        p.sensitive == Some(true)
            && p.kind
                .as_ref()
                .is_some_and(|k| k.is(&zenkey::slice::ProcedureKind::Write))
    })
}

fn procedure_exprs(
    base: &str,
    slices: &SliceSet,
    keep: impl Fn(&zenkey::slice::ProcedureDecl) -> bool,
) -> Vec<String> {
    let mut out = BTreeSet::new();
    for slice in slices.slices() {
        let origin = slice
            .service_origin
            .as_ref()
            .map_or("*".to_string(), |o| o.token().to_string());
        for p in &slice.procedures {
            if keep(p) {
                out.insert(wire(
                    base,
                    format!(
                        "v1/{origin}/@rpc/{}/{}",
                        slice.name,
                        procedure_pattern(&p.path)
                    ),
                ));
            }
        }
    }
    out.into_iter().collect()
}

// ── The principals ────────────────────────────────────────────────────────

/// How a principal names itself in a refusal.
fn principal_name(index: usize, p: &PrincipalSpec) -> String {
    p.id.clone()
        .or_else(|| p.cn.clone())
        .or_else(|| p.zid.clone())
        .unwrap_or_else(|| format!("principal #{}", index + 1))
}

/// A host's origin: given, computed from its machine-id, or both in
/// agreement (RFC 06 §1).
fn host_origin(
    p: &PrincipalSpec,
    salt: Option<&str>,
) -> std::result::Result<String, (String, &'static str)> {
    let given = p.origin.as_deref();
    if let Some(o) = given
        && !zenkey::grammar::is_valid_host_origin(o)
    {
        return Err((
            format!("origin {o:?} is not an `h-<12 hex>` host origin"),
            "RFC 06 §1",
        ));
    }
    let Some(machine_id) = p.machine_id.as_deref() else {
        return given.map(str::to_string).ok_or((
            "a host needs `origin` or `machine_id` — nothing in the grammar binds a \
             transport identity to an origin, so the enrollment has to"
                .to_string(),
            "RFC 03 §4 D6",
        ));
    };
    let trimmed = machine_id.trim();
    if trimmed.len() != 32 || !trimmed.chars().all(|c| c.is_ascii_hexdigit()) {
        return Err((
            format!("machine_id {machine_id:?} is not the 32 hex chars of /etc/machine-id"),
            "RFC 06 §1",
        ));
    }
    let Some(salt) = salt else {
        return Err((
            "machine_id needs the application's origin salt: set `[fleet] salt`".to_string(),
            "RFC 06 §1",
        ));
    };
    // The salt is an application *constant* by design (RFC 06 §1 "salt
    // scope"), and `OriginSalt` says so in its type. The enrollment's stands
    // in for one, and leaking it is what makes a runtime string into that
    // constant — once per plan, a few bytes, in a process that plans once.
    let salt: &'static str = Box::leak(salt.to_string().into_boxed_str());
    let computed = zenkey::origin::HostId::from_machine_id(trimmed, zenkey::OriginSalt::new(salt));
    let computed = computed.as_str().to_string();
    match given {
        Some(o) if o != computed => Err((
            format!(
                "origin {o:?} disagrees with the origin computed from machine_id under this \
                 salt, {computed:?} — one of the three is wrong, and a rule pinned to the \
                 wrong origin fails silently"
            ),
            "RFC 06 §1",
        )),
        _ => Ok(computed),
    }
}

/// A service principal's origin: the default for its role, or a verbatim
/// `@…` chunk it names.
fn service_origin(
    p: &PrincipalSpec,
    default: &str,
) -> std::result::Result<String, (String, &'static str)> {
    match p.origin.as_deref() {
        None => Ok(default.to_string()),
        Some(o) if zenkey::grammar::is_valid_verbatim_chunk(o) => Ok(o.to_string()),
        Some(o) => Err((
            format!("origin {o:?} is not a verbatim `@…` service origin"),
            "RFC 03 §1.2",
        )),
    }
}

fn warn(
    kind: AclWarningKind,
    principal: Option<&str>,
    cite: &str,
    text: impl Into<String>,
) -> AclWarning {
    AclWarning {
        kind,
        principal: principal.map(str::to_string),
        text: text.into(),
        cite: cite.into(),
    }
}

/// Rules keyed by id, in first-insertion order; a second identical push is
/// a no-op (two CNs enrolled for one origin share its rules).
#[derive(Default)]
struct Rules {
    order: Vec<String>,
    by_id: BTreeMap<String, AclRule>,
}

impl Rules {
    fn push(&mut self, r: AclRule) -> String {
        let id = r.id.clone();
        if !self.by_id.contains_key(&id) {
            self.order.push(id.clone());
            self.by_id.insert(id.clone(), r);
        }
        id
    }

    fn into_vec(self) -> Vec<AclRule> {
        let mut by_id = self.by_id;
        self.order
            .into_iter()
            .filter_map(|id| by_id.remove(&id))
            .collect()
    }
}

// ── The plan ──────────────────────────────────────────────────────────────

/// Plan the `access_control` block for an enrollment under `base`.
///
/// `slices`, when given, **narrows** the plan to what the fleet serves: a
/// host's `@media` and `@blob` rules exist only where some host producer
/// declares the plane, and `no-remote-actions` denies exactly the declared
/// `kind = "write"` procedures. Without one, the planes are what the
/// enrollment claims and the deny is the convention's `set` leaf — and the
/// plan says so (RFC 13 §3 O4).
///
/// A refused principal is left out of `subjects` and `policies` and named
/// in `refusals`; the plan is still emitted around it.
pub fn plan_acl(
    enrollment: &Enrollment,
    base: &str,
    slices: Option<&SliceSet>,
    opts: AclOptions,
) -> AclPlan {
    let mut rules = Rules::default();
    let mut subjects = Vec::new();
    let mut policies = Vec::new();
    let mut warnings = Vec::new();
    let mut refusals = Vec::new();

    let facts = slices.map(registry_facts);
    let catalog_adv = enrollment.fleet.catalog_adv;
    let salt = enrollment.fleet.salt.as_deref();

    // What the fleet serves, for the consumers' plane lists: without a
    // registry, everything the enrollment's hosts claim.
    let claimed_media = enrollment
        .principal
        .iter()
        .any(|p| p.role == Role::Host && p.media);
    let planes = match &facts {
        Some(f) => Planes {
            media: !f.media_producers.is_empty(),
            // `@blob` is served by any declarer; seeding is one host's
            // choice on top.
            blob: !f.blob_producers.is_empty(),
            catalog_adv,
        },
        None => Planes {
            media: claimed_media,
            // Without a registry the RFC row stands: every host serves
            // `@blob` (RFC 07 §2 makes it a plane every host may serve).
            blob: true,
            catalog_adv,
        },
    };

    // The write set the consumers are denied.
    let write_set: Vec<String> = match slices {
        Some(s) => {
            let exprs = write_exprs(base, s);
            if exprs.is_empty() {
                warnings.push(warn(
                    AclWarningKind::NoWriteProcedures,
                    None,
                    "RFC 09 §3",
                    "the registry declares no `kind = \"write\"` procedure, so the \
                     no-remote-actions deny is omitted — zenoh refuses a rule with no \
                     key expression",
                ));
            }
            exprs
        }
        None => {
            warnings.push(warn(
                AclWarningKind::WriteSetNotNarrowed,
                None,
                "RFC 13 §3 O4",
                format!(
                    "no registry asked: the planes are as the enrollment claims, and \
                     no-remote-actions denies the convention's write leaf {} rather than \
                     the declared write procedures — pass --registry <dir> to narrow both \
                     to what the fleet actually declares",
                    wire(base, UNNARROWED_WRITE_LEAF)
                ),
            ));
            vec![wire(base, UNNARROWED_WRITE_LEAF)]
        }
    };

    // The sensitive writes (RFC 08 §2, v1.43): denied to every principal,
    // `remote_actions` or not, until a grant names one. Unknown without a
    // registry, and an unknown set denies nothing extra — the unnarrowed
    // leaf already denies every write.
    let sensitive_set: Vec<String> = slices.map(|s| sensitive_exprs(base, s)).unwrap_or_default();
    let narrowed = slices.is_some();

    let interest_prop = || {
        rule(
            INTEREST_PROP,
            AclPermission::Allow,
            Some(OUT),
            &DECLARES,
            fleet_exprs(base, &planes),
            "interest-prop",
            "the fifth fact (reference deployment 2026-08-30): zenoh checks a consumer's \
             declares and queries on egress toward the responding face against that \
             face's subject, and own-origin grants include no wildcard-origin selector \
             — without this a peer-mode publisher publishes to nobody. Egress-only, \
             declares and queries only: puts and replies stay ingress-checked",
        )
    };

    let mut seen_ids = BTreeSet::new();
    let mut seen_cns = BTreeSet::new();
    let mut seen_zids = BTreeSet::new();
    let mut hosts = 0usize;
    let mut consumers = 0usize;

    for (i, p) in enrollment.principal.iter().enumerate() {
        let name = principal_name(i, p);
        let mut refuse = |reason: String, cite: &str| {
            refusals.push(AclRefusal {
                principal: name.clone(),
                reason,
                cite: cite.into(),
            });
        };

        // The subject: who this is on the wire.
        let cn = p.cn.as_deref().filter(|s| !s.is_empty());
        let zid = p.zid.as_deref().filter(|s| !s.is_empty());
        if cn.is_none() && zid.is_none() {
            refuse(
                "a principal needs a `cn` (the certificate common name) — that is the \
                 enrollment"
                    .into(),
                "RFC 03 §4 D6",
            );
            continue;
        }
        if let Some(z) = zid
            && !opts.allow_zid_subjects
        {
            refuse(
                format!(
                    "zid {z:?} as a subject is refused: a ZID is not backed by an \
                     authentication mechanism and can only be trusted for ACL when a \
                     dedicated zenoh mechanism manages it (zenoh's own config says so); \
                     pass --allow-zid-subjects for a prototype, never a deployment"
                ),
                "zenoh-1.10.0/DEFAULT_CONFIG.json5",
            );
            continue;
        }
        if let Some(c) = cn
            && !seen_cns.insert(c.to_string())
        {
            refuse(
                format!("cn {c:?} is already enrolled — one transport identity, one principal"),
                "RFC 03 §4 D6",
            );
            continue;
        }
        if let Some(z) = zid
            && !seen_zids.insert(z.to_string())
        {
            refuse(format!("zid {z:?} is already enrolled"), "RFC 03 §4 D6");
            continue;
        }
        let sid =
            p.id.clone()
                .or_else(|| cn.map(str::to_string))
                .or_else(|| zid.map(str::to_string))
                .expect("cn or zid is present");
        if !seen_ids.insert(sid.clone()) {
            refuse(
                format!("subject id {sid:?} is already taken — give this principal an `id`"),
                "zenoh-config: subject ids are unique",
            );
            continue;
        }
        if let Some(z) = zid {
            warnings.push(warn(
                AclWarningKind::ZidSubject,
                Some(&sid),
                "zenoh-1.10.0/DEFAULT_CONFIG.json5",
                format!(
                    "subject {sid:?} is bound by zid {z:?}: prototyping only — a ZID is not \
                     backed by authentication and a peer can present any"
                ),
            ));
        }
        if p.remote_actions && p.role == Role::Watch {
            refuse(
                "a watch is read-only by definition; `remote_actions = true` wants a \
                 console"
                    .into(),
                "RFC 09 §3",
            );
            continue;
        }
        if !p.writes.is_empty() && p.role != Role::Console {
            refuse(
                format!(
                    "`writes` grants belong to a console; a {} {}",
                    p.role.as_str(),
                    if p.role == Role::Watch {
                        "is read-only by definition"
                    } else {
                        "acts as itself and needs no grant"
                    }
                ),
                "RFC 09 §3",
            );
            continue;
        }
        if p.role == Role::Link {
            refuse(
                "a link is a face, not a principal: it is selected by its transport and \
                 planned by `acl gen --face` from the registry's exposure markers"
                    .into(),
                "RFC 09 §4",
            );
            continue;
        }

        // The role's rules.
        let mut ids: Vec<String> = Vec::new();
        match p.role {
            Role::Host => {
                let origin = match host_origin(p, salt) {
                    Ok(o) => o,
                    Err((reason, cite)) => {
                        refuse(reason, cite);
                        continue;
                    }
                };
                hosts += 1;
                let o = origin.as_str();
                ids.push(rules.push(rule(
                    format!("host-data-{o}"),
                    AclPermission::Allow,
                    Some(IN),
                    &[
                        AclMessage::Put,
                        AclMessage::Delete,
                        AclMessage::LivelinessToken,
                    ],
                    vec![wire(base, format!("v1/{o}/**"))],
                    "host-data",
                    "RFC 09 §3 fact 1: the data classes and the alive tokens — and nothing \
                     under @rpc, @media or @blob, which `**` never reaches",
                )));
                // `@blob` rides host-serve where the plane is served at all.
                let mut serve = vec![wire(base, format!("v1/{o}/@rpc/**"))];
                if planes.blob {
                    serve.push(wire(base, format!("v1/{o}/@blob/**")));
                }
                ids.push(rules.push(rule(
                    format!("host-serve-{o}"),
                    AclPermission::Allow,
                    None,
                    &[
                        AclMessage::DeclareQueryable,
                        AclMessage::Reply,
                        AclMessage::Query,
                    ],
                    serve,
                    "host-serve",
                    "RFC 09 §3 facts 1 and 4: @rpc spelled explicitly because `**` will not \
                     cross it; declare_queryable allowed or the host serves nothing; query \
                     egress is the router forwarding calls to it",
                )));
                if p.media {
                    if planes.media {
                        ids.push(rules.push(rule(
                            format!("host-media-{o}"),
                            AclPermission::Allow,
                            Some(IN),
                            &[AclMessage::Put],
                            vec![wire(base, format!("v1/{o}/@media/**"))],
                            "host-media",
                            "RFC 09 §3 fact 1: the plane needs its own rule",
                        )));
                    } else {
                        warnings.push(warn(
                            AclWarningKind::PlaneNotDeclared,
                            Some(&sid),
                            "RFC 08 §2",
                            "media = true, but no host producer in the registry declares \
                             [[media]] — host-media omitted; the enrollment claims a plane \
                             the fleet does not serve",
                        ));
                    }
                }
                if p.blob_seed {
                    if planes.blob {
                        ids.push(rules.push(rule(
                            format!("host-blob-seed-{o}"),
                            AclPermission::Allow,
                            Some(IN),
                            &[AclMessage::Put],
                            vec![
                                wire(base, format!("v1/{o}/@blob/store/**")),
                                wire(base, format!("v1/{o}/@blob/tree/**")),
                            ],
                            "host-blob-seed",
                            "RFC 09 §3, RFC 07 §2: the sanctioned one-shot PUT path into the \
                             router content store — omit in deployments without one",
                        )));
                    } else {
                        warnings.push(warn(
                            AclWarningKind::PlaneNotDeclared,
                            Some(&sid),
                            "RFC 08 §2",
                            "blob_seed = true, but no host producer in the registry declares \
                             [[blob]] — host-blob-seed omitted",
                        ));
                    }
                }
                if p.adv {
                    ids.push(rules.push(rule(
                        format!("host-adv-{o}"),
                        AclPermission::Allow,
                        None,
                        &[
                            AclMessage::Put,
                            AclMessage::LivelinessToken,
                            AclMessage::DeclareQueryable,
                            AclMessage::Reply,
                            AclMessage::Query,
                        ],
                        vec![wire(base, format!("v1/{o}/**/@adv/**"))],
                        "host-adv",
                        "RFC 09 §3, RFC 04 §3.3: the sidecars live under a verbatim @adv \
                         suffix host-data's `**` cannot reach — omitting this when the tier \
                         is in use fails silently: empty seeds, dead recovery",
                    )));
                }
                ids.push(rules.push(interest_prop()));
            }
            Role::Catalog => {
                let origin = match service_origin(p, zenkey::grammar::SERVICE_CATALOG) {
                    Ok(o) => o,
                    Err((reason, cite)) => {
                        refuse(reason, cite);
                        continue;
                    }
                };
                consumers += 1;
                let o = origin.as_str();
                let c = id_chunk(o);
                let mut own = vec![
                    wire(base, format!("v1/{o}/**")),
                    wire(base, format!("v1/{o}/@rpc/**")),
                ];
                if catalog_adv {
                    own.push(wire(base, format!("v1/{o}/**/@adv/**")));
                }
                ids.push(rules.push(rule(
                    format!("catalog-own-{c}"),
                    AclPermission::Allow,
                    None,
                    &[
                        AclMessage::Put,
                        AclMessage::Delete,
                        AclMessage::LivelinessToken,
                        AclMessage::DeclareQueryable,
                        AclMessage::Reply,
                        AclMessage::Query,
                    ],
                    own,
                    "catalog-own",
                    "RFC 09 §3 fact 2: `*` never covers a verbatim origin, so the catalog's \
                     own keys are their own rule — @rpc and, on the advanced tier, @adv \
                     spelled beside them (fact 1)",
                )));
                let intake = if catalog_adv {
                    vec![wire(base, "v1/**"), wire(base, "v1/**/@adv/**")]
                } else {
                    vec![wire(base, "v1/**")]
                };
                ids.push(rules.push(rule(
                    format!("catalog-intake-declare-{c}"),
                    AclPermission::Allow,
                    Some(IN),
                    &DECLARES,
                    intake.clone(),
                    "catalog-intake-declare",
                    "RFC 09 §3 fact 4: the catalog DECLARES interest on ingress; split by flow \
                     from the receive half so it cannot ingress-publish any host's keys",
                )));
                ids.push(rules.push(rule(
                    format!("catalog-intake-recv-{c}"),
                    AclPermission::Allow,
                    Some(OUT),
                    &RECEIVES,
                    intake,
                    "catalog-intake-recv",
                    "RFC 09 §3: the catalog RECEIVES data and tokens on egress — a flowless \
                     rule here would let it tombstone any host's keys, defeating the \
                     per-host enrollment",
                )));
                ids.push(rules.push(interest_prop()));
            }
            Role::Console => {
                consumers += 1;
                ids.push(rules.push(rule(
                    "ops-sub",
                    AclPermission::Allow,
                    Some(IN),
                    &DECLARES,
                    fleet_exprs(base, &planes),
                    "ops-sub",
                    "RFC 09 §3 fact 4: every plane named, because a declare that is not \
                     allowed receives nothing and `**` reaches no verbatim plane (fact 1)",
                )));
                ids.push(rules.push(rule(
                    "ops-recv",
                    AclPermission::Allow,
                    Some(OUT),
                    &RECEIVES,
                    fleet_exprs(base, &planes),
                    "ops-recv",
                    "RFC 09 §3: what the console receives, on egress, over the same planes",
                )));
                if p.adv {
                    ids.push(rules.push(rule(
                        "ops-own-token",
                        AclPermission::Allow,
                        Some(IN),
                        &[AclMessage::LivelinessToken],
                        adv_exprs(base, catalog_adv),
                        "ops-own-token",
                        "RFC 09 §3: the console's own subscriber-detection token, confined to \
                         @adv — a broad ingress liveliness_token allow would let it forge \
                         any host's state/*/alive roster entry",
                    )));
                }
                for r in write_rules(
                    base,
                    &sid,
                    p,
                    &write_set,
                    &sensitive_set,
                    narrowed,
                    &mut warnings,
                ) {
                    ids.push(rules.push(r));
                }
            }
            Role::Watch => {
                consumers += 1;
                let read_only = Planes {
                    media: false,
                    blob: false,
                    catalog_adv,
                };
                ids.push(rules.push(rule(
                    "watch-sub",
                    AclPermission::Allow,
                    Some(IN),
                    &DECLARES,
                    fleet_exprs(base, &read_only),
                    "watch-sub",
                    "RFC 09 §3 fact 4, for a read-only observer: the data classes, the \
                     catalog and RPC reads — never @media, never @blob",
                )));
                ids.push(rules.push(rule(
                    "watch-recv",
                    AclPermission::Allow,
                    Some(OUT),
                    &RECEIVES,
                    fleet_exprs(base, &read_only),
                    "watch-recv",
                    "RFC 09 §3: what a read-only observer receives, on egress",
                )));
                if p.adv {
                    ids.push(rules.push(rule(
                        "ops-own-token",
                        AclPermission::Allow,
                        Some(IN),
                        &[AclMessage::LivelinessToken],
                        adv_exprs(base, catalog_adv),
                        "ops-own-token",
                        "RFC 09 §3: the observer's own subscriber-detection token, confined \
                         to @adv",
                    )));
                }
                for r in write_rules(
                    base,
                    &sid,
                    p,
                    &write_set,
                    &sensitive_set,
                    narrowed,
                    &mut warnings,
                ) {
                    ids.push(rules.push(r));
                }
            }
            Role::Link => unreachable!("refused above"),
            Role::DesiredAuthor => {
                let origin = match service_origin(p, "@desired") {
                    Ok(o) => o,
                    Err((reason, cite)) => {
                        refuse(reason, cite);
                        continue;
                    }
                };
                let o = origin.as_str();
                ids.push(rules.push(rule(
                    format!("desired-author-{}", id_chunk(o)),
                    AclPermission::Allow,
                    Some(IN),
                    &[AclMessage::Put, AclMessage::Delete],
                    vec![wire(base, format!("v1/{o}/state/**"))],
                    "desired-author",
                    "RFC 09 §3, RFC 07 §3: exactly one ingress put/delete grant, on its OWN \
                     service-origin subtree — never on a host origin; the target host id is \
                     the first subject chunk, not the origin, so it cannot forge a host's \
                     state",
                )));
                ids.push(rules.push(interest_prop()));
            }
        }

        subjects.push(AclSubject {
            id: sid.clone(),
            role: p.role,
            cert_common_names: cn.map(|c| vec![c.to_string()]).unwrap_or_default(),
            zids: zid.map(|z| vec![z.to_string()]).unwrap_or_default(),
            link_protocols: Vec::new(),
            interfaces: Vec::new(),
        });
        policies.push(AclPolicy {
            id: sid.clone(),
            rules: ids,
            subjects: vec![sid],
        });
    }

    if hosts == 0 {
        warnings.push(warn(
            AclWarningKind::RoleAbsent,
            None,
            "RFC 09 §3",
            "no host enrolled: nothing in this fleet may publish an origin",
        ));
    }
    if consumers == 0 {
        warnings.push(warn(
            AclWarningKind::RoleAbsent,
            None,
            "RFC 09 §3 fact 4",
            "no catalog, console or watch enrolled: nothing may declare interest, so \
             every host publishes to nobody",
        ));
    }

    AclPlan {
        base: base.to_string(),
        default_permission: AclPermission::Deny,
        registry: facts.map_or(Asked::NotAsked, Asked::Asked),
        rules: rules.into_vec(),
        subjects,
        policies,
        downsampling: Vec::new(),
        warnings,
        refusals,
    }
}

/// One consumer's write rules (RFC 09 §3, v1.43): its `writes` grants, and
/// its deny — the shared `no-remote-actions` when nothing carves it, a
/// per-principal deny when a grant or `remote_actions` does.
///
/// Deny wins in zenoh, so a grant cannot punch a hole in a deny: the deny
/// has to be the declared writes *minus* what the grants include, which is
/// why a grant needs the registry — the unnarrowed leaf has nothing to
/// subtract from. A `sensitive` write stays in the deny for every
/// principal, `remote_actions` or not, unless a grant includes it.
fn write_rules(
    base: &str,
    sid: &str,
    p: &PrincipalSpec,
    write_set: &[String],
    sensitive_set: &[String],
    narrowed: bool,
    warnings: &mut Vec<AclWarning>,
) -> Vec<AclRule> {
    let mut out = Vec::new();
    let mut grants: Vec<String> = Vec::new();
    if !p.writes.is_empty() {
        if narrowed {
            for pattern in &p.writes {
                let expr = wire(base, format!("v1/*/@rpc/{pattern}"));
                let Ok(ke) = keyexpr::new(expr.as_str()) else {
                    warnings.push(warn(
                        AclWarningKind::GrantMatchesNothing,
                        Some(sid),
                        "RFC 03 §2",
                        format!(
                            "writes pattern {pattern:?} is not a valid key expression — skipped"
                        ),
                    ));
                    continue;
                };
                let covers = write_set
                    .iter()
                    .filter(|w| keyexpr::new(w.as_str()).is_ok_and(|wk| ke.includes(wk)))
                    .count();
                if covers == 0 {
                    warnings.push(warn(
                        AclWarningKind::GrantMatchesNothing,
                        Some(sid),
                        "RFC 09 §3",
                        format!(
                            "writes pattern {pattern:?} includes no declared write procedure — \
                             allowed as spelled, but nothing the registry knows answers there"
                        ),
                    ));
                }
                grants.push(expr);
            }
        } else {
            warnings.push(warn(
                AclWarningKind::GrantNotNarrowed,
                Some(sid),
                "RFC 09 §3",
                format!(
                    "writes = {:?} needs the registry: deny wins, so a grant is a carve-out of \
                     the deny, and without a registry the deny is the unnarrowed leaf with \
                     nothing to subtract — no grant emitted, the deny stays whole; pass \
                     --registry <dir>",
                    p.writes
                ),
            ));
        }
    }
    let granted = |w: &String| {
        grants.iter().any(|g| {
            keyexpr::new(g.as_str())
                .is_ok_and(|gk| keyexpr::new(w.as_str()).is_ok_and(|wk| gk.includes(wk)))
        })
    };
    if !grants.is_empty() {
        out.push(rule(
            format!("writes-{sid}"),
            AclPermission::Allow,
            None,
            &[AclMessage::Query],
            grants.clone(),
            "writes",
            "RFC 09 §3 (v1.43): a per-resource write grant — the resource is a path chunk, \
             so the grant is a literal prefix; the deny below is carved around it, \
             because deny wins and a carve-out has to be an absence",
        ));
    }
    let deny: Vec<String> = if p.remote_actions {
        sensitive_set
            .iter()
            .filter(|w| !granted(w))
            .cloned()
            .collect()
    } else {
        write_set.iter().filter(|w| !granted(w)).cloned().collect()
    };
    if deny.is_empty() {
        return out;
    }
    if deny == write_set {
        out.push(rule(
            NO_REMOTE_ACTIONS,
            AclPermission::Deny,
            None,
            &[AclMessage::Query],
            deny,
            "no-remote-actions",
            "RFC 09 §3: the write procedures are deniable per key because the key \
             IS the target; deny wins for a caller who spells it, and fact 6 \
             leaves a broader query to the server (RFC 05 §2.1)",
        ));
    } else {
        out.push(rule(
            format!("{NO_REMOTE_ACTIONS}-{sid}"),
            AclPermission::Deny,
            None,
            &[AclMessage::Query],
            deny,
            "no-remote-actions",
            if p.remote_actions {
                "RFC 08 §2 (v1.43): the sensitive writes stay denied to a console that may \
                 act — RFC 8341's default-deny-all as data — until a `writes` grant names one"
            } else {
                "RFC 09 §3 (v1.43): the declared writes this principal's grants do not \
                 include — carved, because deny wins and a grant cannot punch a hole"
            },
        ));
    }
    out
}

// ── The constrained face ──────────────────────────────────────────────────

const BOTH: &[AclFlow] = &[AclFlow::Egress, AclFlow::Ingress];

/// What a data class can carry over a face — the message kinds a deny on
/// it must name, both flows, because a rule that names neither relies on
/// a default the next zenoh may read differently.
fn class_messages(class: &str) -> &'static [AclMessage] {
    match class {
        "state" => &AclMessage::ALL,
        _ => &[
            AclMessage::Put,
            AclMessage::Delete,
            AclMessage::DeclareSubscriber,
            AclMessage::Query,
            AclMessage::Reply,
            AclMessage::DeclareQueryable,
        ],
    }
}

/// Plan one constrained face (RFC 09 §4, v1.43) from the registry's
/// `exposure` markers: the `access_control` block that keeps `host`
/// surfaces off the link, the `downsampling` block that caps the `link`
/// ones, and a note naming what still crosses.
///
/// Two facts shape the block, both found the hard way by the reference
/// adopter: `default_permission` is node-global, so a face-scoped deny
/// lives under `allow` and the host bus keeps flowing by absence; and deny
/// is by inclusion (RFC 09 §3 fact 6), so a *plane* is denied by its widest
/// spelling (`**/@rpc/**`), never by a producer-scoped literal a broader
/// query would step over. The default exposure is `fleet`, so an unmarked
/// registry crosses whole — said in a warning, never denied on the
/// planner's initiative: the registry is where the classification belongs.
pub fn plan_face(slices: &SliceSet, base: &str, spec: &FaceSpec) -> AclPlan {
    use zenkey::slice::{Declared, Exposure};

    let mut rules = Rules::default();
    let mut ids: Vec<String> = Vec::new();
    let mut warnings = Vec::new();
    let mut downsampling: Vec<AclDownsample> = Vec::new();

    // A `link` entry is affordable only when the face states a rate; with
    // none, it is kept home like a `host` one.
    let effective = |d: Option<&Declared<Exposure>>| -> Exposure {
        match d {
            Some(Declared::Known(Exposure::Host)) => Exposure::Host,
            Some(Declared::Known(Exposure::Link)) => match spec.interval {
                LinkInterval::None => Exposure::Host,
                LinkInterval::EverySecs(_) => Exposure::Link,
            },
            // Unmarked, or a token this build cannot read: it crosses, and
            // the note below says so.
            _ => Exposure::Fleet,
        }
    };

    let mut sorted: Vec<&zenkey::slice::RegistrySlice> = slices.slices().iter().collect();
    sorted.sort_by(|a, b| a.name.cmp(&b.name));

    let mut any_host_procedure = false;
    let mut cut_procedures: Vec<String> = Vec::new();
    let mut open_procedures: Vec<String> = Vec::new();

    for slice in &sorted {
        let origin = slice
            .service_origin
            .as_ref()
            .map_or("*".to_string(), |o| o.token().to_string());
        let producer = slice.name.as_str();
        let mut by_class: BTreeMap<String, Vec<&zenkey::slice::SubjectDecl>> = BTreeMap::new();
        for s in &slice.subjects {
            by_class
                .entry(s.class.token().to_string())
                .or_default()
                .push(s);
        }
        let mut crosses: Vec<String> = Vec::new();
        for (class, subjects) in &by_class {
            let mut kept: Vec<String> = Vec::new();
            for s in subjects {
                let key = wire(
                    base,
                    format!(
                        "v1/{origin}/{class}/{producer}/{}",
                        procedure_pattern(&s.path)
                    ),
                );
                match effective(s.exposure.as_ref()) {
                    Exposure::Host => kept.push(key),
                    Exposure::Link => {
                        if let LinkInterval::EverySecs(secs) = spec.interval {
                            downsampling.push(AclDownsample {
                                key_expr: key,
                                freq: 1.0 / secs.max(1) as f64,
                            });
                        }
                    }
                    Exposure::Fleet => crosses.push(format!("{class}/{}", s.path)),
                }
            }
            if kept.is_empty() {
                continue;
            }
            if kept.len() == subjects.len() {
                // Every subject of the class stays home: one deny on the
                // subtree, which also covers the framework keys under it.
                ids.push(rules.push(rule(
                    format!("deny-{class}-{producer}"),
                    AclPermission::Deny,
                    Some(BOTH),
                    class_messages(class),
                    vec![wire(base, format!("v1/{origin}/{class}/{producer}/**"))],
                    "face-deny-class",
                    "RFC 09 §4 (v1.43): every subject of this class is `host`, so the class \
                     stays home whole — the subtree, not a list that a new subject would \
                     fall outside of",
                )));
            } else {
                ids.push(rules.push(rule(
                    format!("deny-{class}-{producer}-detail"),
                    AclPermission::Deny,
                    Some(BOTH),
                    class_messages(class),
                    kept,
                    "face-deny-detail",
                    "RFC 09 §4 (v1.43): the `host` subjects by name, because a sibling is \
                     `link` or `fleet` and deny is evaluated first — the carve-out for what \
                     crosses has to be an absence from this list",
                )));
            }
        }
        for p in &slice.procedures {
            let name = format!("{producer}/{}", p.path);
            match effective(p.exposure.as_ref()) {
                Exposure::Host => any_host_procedure = true,
                Exposure::Link | Exposure::Fleet => {
                    cut_procedures.push(name.clone());
                    open_procedures.push(name);
                }
            }
        }
        if !crosses.is_empty() {
            warnings.push(warn(
                AclWarningKind::FaceCrosses,
                Some(producer),
                "RFC 08 §2",
                format!(
                    "{} subject(s) cross this face as `fleet` (declared, or unmarked — the \
                     default): {}",
                    crosses.len(),
                    crosses.join(", ")
                ),
            ));
        }
    }

    // The planes, by their widest spelling (fact 6). @rpc when any
    // procedure stays home; @media and @blob always — a stream or a bulk
    // fetch has no business on a constrained link (RFC 09 §4); the admin
    // space and the sidecars always.
    if any_host_procedure {
        ids.push(rules.push(rule(
            "deny-rpc",
            AclPermission::Deny,
            Some(BOTH),
            &[
                AclMessage::Query,
                AclMessage::Reply,
                AclMessage::DeclareQueryable,
                AclMessage::Put,
                AclMessage::Delete,
                AclMessage::DeclareSubscriber,
            ],
            vec!["**/@rpc/**".to_string()],
            "face-deny-plane",
            "RFC 09 §3 fact 6 (v1.38): a plane is denied by the widest pattern that can \
             reach it — the verbatim chunk must be spelled by any query that reaches the \
             plane, and `**` on either side includes every spelling around it; a \
             producer-scoped literal is crossed by a broader query",
        )));
        if !cut_procedures.is_empty() {
            warnings.push(warn(
                AclWarningKind::FaceCutsPlane,
                None,
                "RFC 09 §3 fact 6",
                format!(
                    "{} procedure(s) marked `link` or `fleet` are cut with the @rpc plane, \
                     which is denied whole because a sibling is `host`: {}",
                    cut_procedures.len(),
                    cut_procedures.join(", ")
                ),
            ));
        }
    } else if !open_procedures.is_empty() {
        warnings.push(warn(
            AclWarningKind::FaceCrosses,
            None,
            "RFC 08 §2",
            format!(
                "the @rpc plane crosses this face: no procedure is marked `host`, so \
                 {} procedure(s) answer over the link: {}",
                open_procedures.len(),
                open_procedures.join(", ")
            ),
        ));
    }
    for (id, plane) in [("deny-media", "@media"), ("deny-blob", "@blob")] {
        ids.push(rules.push(rule(
            id,
            AclPermission::Deny,
            Some(BOTH),
            &[
                AclMessage::Put,
                AclMessage::Delete,
                AclMessage::DeclareSubscriber,
                AclMessage::Query,
                AclMessage::Reply,
                AclMessage::DeclareQueryable,
            ],
            vec![format!("**/{plane}/**")],
            "face-deny-plane",
            "RFC 09 §4: a stream or a bulk fetch has no business on a constrained link; \
             the widest spelling, for fact 6",
        )));
    }
    ids.push(rules.push(rule(
        "deny-adminspace",
        AclPermission::Deny,
        Some(BOTH),
        &AclMessage::ALL,
        vec!["@/**".to_string(), "**/@adv/**".to_string()],
        "face-deny-admin",
        "RFC 09 §4: the admin space and the advanced-tier sidecars never cross a \
         constrained link — every rule above is rooted at v1, so without this the \
         whole admin surface would be permitted by the permissive default",
    )));

    // Sorted by key, so the block reads the same whatever order the
    // registry lists its subjects in, and a diff of two generations is a
    // diff of the markers.
    downsampling.sort_by(|a, b| a.key_expr.cmp(&b.key_expr));
    // One timer per rule in zenoh, and a key resolves to the first
    // intersecting rule: two intersecting rules would share a budget with
    // an arbitrary winner.
    for (i, a) in downsampling.iter().enumerate() {
        for b in &downsampling[i + 1..] {
            if let (Ok(ka), Ok(kb)) = (
                keyexpr::new(a.key_expr.as_str()),
                keyexpr::new(b.key_expr.as_str()),
            ) && ka.intersects(kb)
            {
                warnings.push(warn(
                    AclWarningKind::FaceRulesIntersect,
                    None,
                    "zenoh downsampling: one timer per rule",
                    format!(
                        "{} and {} intersect and would share one downsampling timer with an \
                         arbitrary winner — declare the subjects so their patterns do not",
                        a.key_expr, b.key_expr
                    ),
                ));
            }
        }
    }

    let subject = AclSubject {
        id: spec.id.clone(),
        role: Role::Link,
        cert_common_names: Vec::new(),
        zids: Vec::new(),
        link_protocols: spec.link_protocols.clone(),
        interfaces: spec.interfaces.clone(),
    };
    let policy = AclPolicy {
        id: format!("{}-mgmt", spec.id),
        rules: ids,
        subjects: vec![spec.id.clone()],
    };
    AclPlan {
        base: base.to_string(),
        default_permission: AclPermission::Allow,
        registry: Asked::Asked(registry_facts(slices)),
        rules: rules.into_vec(),
        subjects: vec![subject],
        policies: vec![policy],
        downsampling,
        warnings,
        refusals: Vec::new(),
    }
}

// ── JSON5 ─────────────────────────────────────────────────────────────────

fn js(s: &str) -> String {
    serde_json::to_string(s).expect("a string serializes")
}

fn js_list<'a>(items: impl IntoIterator<Item = &'a str>) -> String {
    let inner: Vec<String> = items.into_iter().map(js).collect();
    format!("[{}]", inner.join(", "))
}

/// The plan as the `access_control` block `zenohd` reads — with a comment
/// per rule naming the matrix row and the fact it exists for.
///
/// Field names are zenoh 1.10's (`zenoh-config-1.10.0/src/lib.rs`:
/// `AclConfig`, `AclConfigRule`, `AclMessage`, `InterceptorFlow`,
/// `AclConfigSubjects`, `AclConfigPolicyEntry`); `flows` is omitted where the
/// rule applies in both directions, which is what zenoh reads an absent
/// `flows` as.
pub fn to_json5(plan: &AclPlan) -> String {
    use std::fmt::Write as _;
    let mut out = String::new();
    let _ = writeln!(
        out,
        "// zenohd access_control block — generated by `zenctl acl gen` (RFC 09 §3)."
    );
    let _ = write!(
        out,
        "// base {}; {} subject(s), {} rule(s), {} polic(y/ies); ",
        js(&plan.base),
        plan.subjects.len(),
        plan.rules.len(),
        plan.policies.len()
    );
    match plan.registry.as_option() {
        Some(r) => {
            let _ = writeln!(
                out,
                "registry: {} slice(s), {} write procedure(s), media by [{}], blob by [{}].",
                r.slices,
                r.write_procedures.len(),
                r.media_producers.join(", "),
                r.blob_producers.join(", ")
            );
        }
        None => {
            let _ = writeln!(
                out,
                "registry: not asked — planes are as the enrollment claims and the write \
                 set is the convention's `set` leaf, unnarrowed."
            );
        }
    }
    let _ = writeln!(
        out,
        "// Field names: zenoh 1.10 (zenoh-config-1.10.0/src/lib.rs — AclConfig, AclConfigRule,\n\
         // AclMessage, InterceptorFlow, AclConfigSubjects, AclConfigPolicyEntry). A rule with no\n\
         // `flows` applies in both directions. Merge at the router config's top level; the\n\
         // three lists are all required — rules alone are refused at startup (fact 3). ACL\n\
         // config is not runtime-reloadable: enrolling a host is a router restart (RFC 03 §4 D6)."
    );
    for r in &plan.refusals {
        let _ = writeln!(out, "// REFUSED {}: {} ({})", r.principal, r.reason, r.cite);
    }
    for w in &plan.warnings {
        let who = w
            .principal
            .as_deref()
            .map_or(String::new(), |p| format!(" [{p}]"));
        let _ = writeln!(
            out,
            "// ! {}{who}: {} ({})",
            w.kind.as_str(),
            w.text,
            w.cite
        );
    }
    let _ = writeln!(out, "access_control: {{");
    let _ = writeln!(out, "  enabled: true,");
    match plan.default_permission {
        AclPermission::Deny => {
            let _ = writeln!(
                out,
                "  default_permission: \"deny\",  // fact 4: deny, and every declaration allowed by name"
            );
        }
        AclPermission::Allow => {
            let _ = writeln!(
                out,
                "  default_permission: \"allow\",  // node-global: a face-scoped deny lives under a permissive default (RFC 09 §4)"
            );
        }
    }

    let _ = writeln!(out, "  rules: [");
    for r in &plan.rules {
        let _ = writeln!(out, "    // {}: {}", r.purpose, r.cite);
        let _ = write!(
            out,
            "    {{ id: {}, permission: {}",
            js(&r.id),
            js(r.permission.as_str())
        );
        if let Some(flows) = &r.flows {
            let _ = write!(
                out,
                ", flows: {}",
                js_list(flows.iter().map(|f| f.as_str()))
            );
        }
        let _ = writeln!(out, ",");
        let _ = writeln!(
            out,
            "      messages: {},",
            js_list(r.messages.iter().map(|m| m.as_str()))
        );
        let _ = writeln!(
            out,
            "      key_exprs: {} }},",
            js_list(r.key_exprs.iter().map(String::as_str))
        );
    }
    let _ = writeln!(out, "  ],");

    let _ = writeln!(
        out,
        "  subjects: [  // the enrollment: transport identity ↔ origin (RFC 03 §4 D6)"
    );
    for s in &plan.subjects {
        let _ = write!(out, "    {{ id: {}", js(&s.id));
        if !s.cert_common_names.is_empty() {
            let _ = write!(
                out,
                ", cert_common_names: {}",
                js_list(s.cert_common_names.iter().map(String::as_str))
            );
        }
        if !s.zids.is_empty() {
            let _ = write!(
                out,
                ", zids: {}",
                js_list(s.zids.iter().map(String::as_str))
            );
        }
        if !s.link_protocols.is_empty() {
            let _ = write!(
                out,
                ", link_protocols: {}",
                js_list(s.link_protocols.iter().map(String::as_str))
            );
        }
        if !s.interfaces.is_empty() {
            let _ = write!(
                out,
                ", interfaces: {}",
                js_list(s.interfaces.iter().map(String::as_str))
            );
        }
        let _ = writeln!(out, " }},  // {}", s.role.as_str());
    }
    let _ = writeln!(out, "  ],");

    let _ = writeln!(out, "  policies: [");
    for p in &plan.policies {
        let _ = writeln!(
            out,
            "    {{ id: {}, rules: {},\n      subjects: {} }},",
            js(&p.id),
            js_list(p.rules.iter().map(String::as_str)),
            js_list(p.subjects.iter().map(String::as_str))
        );
    }
    let _ = writeln!(out, "  ],");
    // Each block ends in a member comma (#486): the output is a fragment
    // pasted into a router config's top-level object, and a trailing comma
    // is legal JSON5 even before the closing `}`.
    let _ = writeln!(out, "}},");
    if !plan.downsampling.is_empty() {
        // The face's second block (RFC 09 §4): egress puts only — dropping
        // on ingress saves no airtime, it has already been spent.
        let face = plan.subjects.first().expect("a face plan has its subject");
        let _ = writeln!(
            out,
            "// `link`-exposed subjects, capped (RFC 09 §4, v1.43): one rule per key, none intersecting —\n\
             // zenoh resolves a key to the first intersecting rule and gives each rule one timer."
        );
        let _ = writeln!(out, "downsampling: [");
        let _ = write!(out, "  {{ id: {}", js(&format!("{}-telemetry", face.id)));
        if !face.link_protocols.is_empty() {
            let _ = write!(
                out,
                ", link_protocols: {}",
                js_list(face.link_protocols.iter().map(String::as_str))
            );
        }
        if !face.interfaces.is_empty() {
            let _ = write!(
                out,
                ", interfaces: {}",
                js_list(face.interfaces.iter().map(String::as_str))
            );
        }
        let _ = writeln!(out, ",");
        let _ = writeln!(out, "    flows: [\"egress\"], messages: [\"put\"],");
        let _ = writeln!(out, "    rules: [");
        for d in &plan.downsampling {
            let _ = writeln!(
                out,
                "      {{ key_expr: {}, freq: {} }},",
                js(&d.key_expr),
                d.freq
            );
        }
        let _ = writeln!(out, "    ] }},");
        let _ = writeln!(out, "],");
    }
    out
}

// ── --check ───────────────────────────────────────────────────────────────

fn set<'a>(items: impl IntoIterator<Item = &'a str>) -> BTreeSet<String> {
    items.into_iter().map(str::to_string).collect()
}

/// A rule's comparable form: flows normalised so that absent = both.
fn rule_shape(
    permission: &str,
    flows: Option<&[String]>,
    messages: &[String],
    key_exprs: &[String],
) -> String {
    let flows: BTreeSet<&str> = match flows {
        Some(f) => f.iter().map(String::as_str).collect(),
        None => ["egress", "ingress"].into_iter().collect(),
    };
    let messages: BTreeSet<&str> = messages.iter().map(String::as_str).collect();
    let key_exprs: BTreeSet<&str> = key_exprs.iter().map(String::as_str).collect();
    format!(
        "{permission} {} {} {}",
        flows.into_iter().collect::<Vec<_>>().join("+"),
        messages.into_iter().collect::<Vec<_>>().join(","),
        key_exprs.into_iter().collect::<Vec<_>>().join(" ")
    )
}

fn planned_rule_shape(r: &AclRule) -> String {
    let flows: Option<Vec<String>> = r
        .flows
        .as_ref()
        .map(|f| f.iter().map(|x| x.as_str().to_string()).collect());
    let messages: Vec<String> = r.messages.iter().map(|m| m.as_str().to_string()).collect();
    rule_shape(
        r.permission.as_str(),
        flows.as_deref(),
        &messages,
        &r.key_exprs,
    )
}

/// The plan against the block a router's config file carries.
///
/// The observed side is what zenoh's own loader parsed, so what is compared
/// is what `zenohd` would run; `against` names the file (RFC 13 §3 O5). The
/// judged claim is *the block differs from the plan* — a finding is
/// `Established`, a block that carries the plan whole is `NotEstablished`.
pub fn check_acl(plan: &AclPlan, observed: &AclConfigDoc, against: &str) -> AclCheck {
    let mut findings = Vec::new();
    let mut finding = |kind, id: &str, planned: Option<String>, observed: Option<String>| {
        findings.push(AclFinding {
            kind,
            id: id.to_string(),
            planned,
            observed,
        });
    };

    if !observed.enabled {
        finding(
            AclFindingKind::Disabled,
            "access_control",
            Some("enabled: true".into()),
            Some("enabled: false".into()),
        );
    }
    if observed.default_permission != plan.default_permission.as_str() {
        finding(
            AclFindingKind::DefaultPermissionDiffers,
            "default_permission",
            Some(plan.default_permission.as_str().into()),
            Some(observed.default_permission.clone()),
        );
    }

    // Rules, by id.
    let observed_rules: BTreeMap<&str, &crate::report::AclRuleDoc> =
        observed.rules.iter().map(|r| (r.id.as_str(), r)).collect();
    for p in &plan.rules {
        let shape = planned_rule_shape(p);
        match observed_rules.get(p.id.as_str()) {
            None => finding(AclFindingKind::RuleMissing, &p.id, Some(shape), None),
            Some(o) => {
                let o_shape =
                    rule_shape(&o.permission, o.flows.as_deref(), &o.messages, &o.key_exprs);
                if o_shape != shape {
                    finding(
                        AclFindingKind::RuleDiffers,
                        &p.id,
                        Some(shape),
                        Some(o_shape),
                    );
                }
            }
        }
    }
    let planned_rule_ids: BTreeSet<&str> = plan.rules.iter().map(|r| r.id.as_str()).collect();
    for o in &observed.rules {
        if !planned_rule_ids.contains(o.id.as_str()) {
            finding(
                AclFindingKind::RuleExtra,
                &o.id,
                None,
                Some(rule_shape(
                    &o.permission,
                    o.flows.as_deref(),
                    &o.messages,
                    &o.key_exprs,
                )),
            );
        }
    }

    // Subjects, by id — and every configured CN against the enrollment.
    let observed_subjects: BTreeMap<&str, &crate::report::AclSubjectDoc> = observed
        .subjects
        .iter()
        .map(|s| (s.id.as_str(), s))
        .collect();
    let known_cns: BTreeSet<&str> = plan
        .subjects
        .iter()
        .flat_map(|s| s.cert_common_names.iter().map(String::as_str))
        .collect();
    let subject_shape = |cns: &[String], zids: &[String], protos: &[String], ifs: &[String]| {
        let mut s = format!("cns {cns:?} zids {zids:?}");
        if !protos.is_empty() {
            s.push_str(&format!(" link_protocols {protos:?}"));
        }
        if !ifs.is_empty() {
            s.push_str(&format!(" interfaces {ifs:?}"));
        }
        s
    };
    for p in &plan.subjects {
        let planned = subject_shape(
            &p.cert_common_names,
            &p.zids,
            &p.link_protocols,
            &p.interfaces,
        );
        match observed_subjects.get(p.id.as_str()) {
            None => finding(AclFindingKind::SubjectMissing, &p.id, Some(planned), None),
            Some(o) => {
                let o_cns = o.cert_common_names.clone().unwrap_or_default();
                let o_zids = o.zids.clone().unwrap_or_default();
                let o_protos = o.link_protocols.clone().unwrap_or_default();
                let o_ifs = o.interfaces.clone().unwrap_or_default();
                let differs = |a: &[String], b: &[String]| {
                    set(a.iter().map(String::as_str)) != set(b.iter().map(String::as_str))
                };
                if differs(&o_cns, &p.cert_common_names)
                    || differs(&o_zids, &p.zids)
                    || differs(&o_protos, &p.link_protocols)
                    || differs(&o_ifs, &p.interfaces)
                {
                    finding(
                        AclFindingKind::SubjectDiffers,
                        &p.id,
                        Some(planned),
                        Some(subject_shape(&o_cns, &o_zids, &o_protos, &o_ifs)),
                    );
                }
            }
        }
    }
    let planned_subject_ids: BTreeSet<&str> = plan.subjects.iter().map(|s| s.id.as_str()).collect();
    for o in &observed.subjects {
        if !planned_subject_ids.contains(o.id.as_str()) {
            finding(
                AclFindingKind::SubjectExtra,
                &o.id,
                None,
                Some(format!(
                    "cns {:?} zids {:?}",
                    o.cert_common_names.clone().unwrap_or_default(),
                    o.zids.clone().unwrap_or_default()
                )),
            );
        }
        // A transport-selected property is planned only on a face (v1.43);
        // on an identity subject it is a property nobody planned.
        let planned_face = plan
            .subjects
            .iter()
            .any(|p| p.id == o.id && (!p.link_protocols.is_empty() || !p.interfaces.is_empty()));
        for (prop, value) in [
            ("interfaces", &o.interfaces),
            ("usernames", &o.usernames),
            ("link_protocols", &o.link_protocols),
        ] {
            if planned_face && prop != "usernames" {
                continue;
            }
            if value.as_ref().is_some_and(|v| !v.is_empty()) {
                finding(
                    AclFindingKind::SubjectUnplannedProperty,
                    &o.id,
                    None,
                    Some(format!("{prop}: {:?}", value.clone().unwrap_or_default())),
                );
            }
        }
        for cn in o.cert_common_names.iter().flatten() {
            if !known_cns.contains(cn.as_str()) {
                finding(
                    AclFindingKind::UnknownCn,
                    cn,
                    None,
                    Some(format!("bound by subject {:?}", o.id)),
                );
            }
        }
    }

    // Policies, as sets of (rules, subjects) — ids are optional in zenoh's
    // shape, so identity is what a policy binds.
    let policy_key = |rules: &[String], subjects: &[String]| {
        format!(
            "rules [{}] subjects [{}]",
            set(rules.iter().map(String::as_str))
                .into_iter()
                .collect::<Vec<_>>()
                .join(", "),
            set(subjects.iter().map(String::as_str))
                .into_iter()
                .collect::<Vec<_>>()
                .join(", ")
        )
    };
    let observed_policies: BTreeSet<String> = observed
        .policies
        .iter()
        .map(|p| policy_key(&p.rules, &p.subjects))
        .collect();
    let planned_policies: BTreeSet<String> = plan
        .policies
        .iter()
        .map(|p| policy_key(&p.rules, &p.subjects))
        .collect();
    for p in &plan.policies {
        let key = policy_key(&p.rules, &p.subjects);
        if !observed_policies.contains(&key) {
            finding(AclFindingKind::PolicyMissing, &p.id, Some(key), None);
        }
    }
    for (i, o) in observed.policies.iter().enumerate() {
        let key = policy_key(&o.rules, &o.subjects);
        if !planned_policies.contains(&key) {
            let id = o.id.clone().unwrap_or_else(|| format!("policy #{}", i + 1));
            finding(AclFindingKind::PolicyExtra, &id, None, Some(key));
        }
    }

    let judgement = if findings.is_empty() {
        Judgement::NotEstablished {
            reason: format!(
                "{against} carries the plan whole: {} rule(s), {} subject(s), {} polic(y/ies), \
                 enabled, default {}",
                plan.rules.len(),
                plan.subjects.len(),
                plan.policies.len(),
                plan.default_permission.as_str()
            ),
        }
    } else {
        Judgement::Established
    };
    AclCheck {
        base: plan.base.clone(),
        against: against.to_string(),
        planned_rules: plan.rules.len(),
        observed_rules: observed.rules.len(),
        planned_subjects: plan.subjects.len(),
        observed_subjects: observed.subjects.len(),
        findings,
        // Observable only from a publisher's matching listener (RFC 07 §1);
        // a config-file check has no publisher to ask, and a consumer-side
        // subscriber cannot see whether its interest was forwarded.
        interest_probe: Judgement::NotAsked,
        judgement,
    }
}

// ── --explain ─────────────────────────────────────────────────────────────

/// Does `principal` hold `message` on `key`, in each direction, and via
/// which rules — over the plan alone, inclusion by `zenoh-keyexpr`.
///
/// `principal` is a subject id, a CN or a zid. A principal the plan does
/// not carry (unknown, or refused) is an [`Error::Unaskable`]: there is no
/// policy to explain.
pub fn explain_acl(
    plan: &AclPlan,
    principal: &str,
    key: &str,
    message: AclMessage,
) -> Result<AclExplain> {
    let subject = plan
        .subjects
        .iter()
        .find(|s| {
            s.id == principal
                || s.cert_common_names.iter().any(|c| c == principal)
                || s.zids.iter().any(|z| z == principal)
        })
        .ok_or_else(|| {
            let refused = plan.refusals.iter().any(|r| r.principal == principal);
            Error::unaskable(
                "explain",
                if refused {
                    format!("principal {principal:?} was refused by the plan — see its refusal")
                } else {
                    format!(
                        "principal {principal:?} is not enrolled; the plan knows {:?}",
                        plan.subjects
                            .iter()
                            .map(|s| s.id.as_str())
                            .collect::<Vec<_>>()
                    )
                },
            )
        })?;
    let k = keyexpr::new(key).map_err(|e| {
        Error::unaskable(
            "explain",
            format!("{key:?} is not a valid key expression (RFC 03 §2): {e}"),
        )
    })?;

    let rule_ids: BTreeSet<&str> = plan
        .policies
        .iter()
        .filter(|p| p.subjects.contains(&subject.id))
        .flat_map(|p| p.rules.iter().map(String::as_str))
        .collect();
    let rules: Vec<&AclRule> = plan
        .rules
        .iter()
        .filter(|r| rule_ids.contains(r.id.as_str()))
        .collect();

    let direction = |flow: AclFlow| -> AclDirection {
        let applicable = rules.iter().filter(|r| {
            r.messages.contains(&message) && r.flows.as_ref().is_none_or(|f| f.contains(&flow))
        });
        let mut via = Vec::new();
        let mut near: Vec<String> = Vec::new();
        for r in applicable {
            let including = r
                .key_exprs
                .iter()
                .find(|e| keyexpr::new(e.as_str()).is_ok_and(|ke| ke.includes(k)));
            match including {
                Some(e) => via.push(AclGrant {
                    rule: r.id.clone(),
                    permission: r.permission,
                    key_expr: e.clone(),
                    purpose: r.purpose.clone(),
                }),
                None => {
                    if let Some(e) = r
                        .key_exprs
                        .iter()
                        .find(|e| keyexpr::new(e.as_str()).is_ok_and(|ke| ke.intersects(k)))
                    {
                        near.push(format!("{} ({e})", r.id));
                    }
                }
            }
        }
        // Deny first, so the reader sees what won.
        via.sort_by_key(|g| g.permission == AclPermission::Allow);
        let denies: Vec<&str> = via
            .iter()
            .filter(|g| g.permission == AclPermission::Deny)
            .map(|g| g.rule.as_str())
            .collect();
        let allows: Vec<&str> = via
            .iter()
            .filter(|g| g.permission == AclPermission::Allow)
            .map(|g| g.rule.as_str())
            .collect();
        let (decision, mut reason) = if !denies.is_empty() {
            (
                AclDecision::Denied,
                format!(
                    "{} denies {} on {} — deny wins{}",
                    denies.join(", "),
                    message.as_str(),
                    flow.as_str(),
                    if allows.is_empty() {
                        String::new()
                    } else {
                        format!(" over {}", allows.join(", "))
                    }
                ),
            )
        } else if !allows.is_empty() {
            (
                AclDecision::Allowed,
                format!(
                    "{} includes it for {} on {}",
                    allows.join(", "),
                    message.as_str(),
                    flow.as_str()
                ),
            )
        } else if plan.default_permission == AclPermission::Allow {
            (
                AclDecision::AllowedByDefault,
                format!(
                    "no rule of {}'s policies includes {key} for {} on {}: this block's \
                     default_permission is allow, so it crosses — a face keeps things home \
                     by deny, and what nothing denies is what the registry marks `fleet` \
                     (RFC 09 §4)",
                    subject.id,
                    message.as_str(),
                    flow.as_str()
                ),
            )
        } else {
            (
                AclDecision::DeniedByDefault,
                format!(
                    "no rule of {}'s policies includes {key} for {} on {}: default_permission \
                     deny",
                    subject.id,
                    message.as_str(),
                    flow.as_str()
                ),
            )
        };
        if decision == AclDecision::DeniedByDefault && !near.is_empty() {
            let verbatim = key.split('/').any(|c| c.starts_with('@'));
            reason.push_str(&format!(
                " — {} intersect{} it but ACL matching is inclusion (RFC 09 §3 fact 1){}",
                near.join(", "),
                if near.len() == 1 { "s" } else { "" },
                if verbatim {
                    ", and `**` never crosses a verbatim `@` chunk (RFC 03 §4 D2): name the \
                     plane in its own rule"
                } else {
                    ""
                }
            ));
        }
        AclDirection {
            decision,
            via,
            reason,
        }
    };

    Ok(AclExplain {
        principal: subject.id.clone(),
        key: key.to_string(),
        message,
        base: plan.base.clone(),
        ingress: direction(AclFlow::Ingress),
        egress: direction(AclFlow::Egress),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::report::{FleetSpec, PrincipalSpec};

    fn principal(cn: &str, role: Role) -> PrincipalSpec {
        PrincipalSpec {
            cn: Some(cn.into()),
            role,
            ..Default::default()
        }
    }

    /// Three hosts, a catalog, a console, a watch — the fixture fleet.
    fn fleet() -> Enrollment {
        Enrollment {
            base: Some("zensight".into()),
            fleet: FleetSpec {
                catalog_adv: true,
                salt: Some("example-salt-v1".into()),
            },
            principal: vec![
                PrincipalSpec {
                    origin: Some("h-3fa9c2d41b7e".into()),
                    adv: true,
                    blob_seed: true,
                    media: true,
                    ..principal("h-3fa9c2d41b7e", Role::Host)
                },
                // RFC 06 §1's test vector: computed, not given.
                PrincipalSpec {
                    machine_id: Some("b642b4217b34b1e8d3bd915fc65c4452".into()),
                    ..principal("edge-02.example", Role::Host)
                },
                PrincipalSpec {
                    origin: Some("h-0123456789ab".into()),
                    adv: true,
                    ..principal("h-0123456789ab", Role::Host)
                },
                principal("zensight-catalog", Role::Catalog),
                PrincipalSpec {
                    adv: true,
                    ..principal("zensight-console", Role::Console)
                },
                principal("zensight-watch", Role::Watch),
            ],
        }
    }

    fn plan() -> AclPlan {
        plan_acl(&fleet(), "zensight", None, AclOptions::default())
    }

    fn rule_of<'p>(plan: &'p AclPlan, id: &str) -> &'p AclRule {
        plan.rules
            .iter()
            .find(|r| r.id == id)
            .unwrap_or_else(|| panic!("rule {id}"))
    }

    fn policy_of<'p>(plan: &'p AclPlan, id: &str) -> &'p AclPolicy {
        plan.policies
            .iter()
            .find(|p| p.id == id)
            .unwrap_or_else(|| panic!("policy {id}"))
    }

    /// Fact 3: three lists, all present; one subject and one policy per
    /// enrolled principal.
    #[test]
    fn the_three_lists_are_present_and_nothing_was_refused() {
        let p = plan();
        assert!(p.refusals.is_empty(), "{:?}", p.refusals);
        assert_eq!(p.subjects.len(), 6);
        assert_eq!(p.policies.len(), 6);
        assert!(!p.rules.is_empty());
        assert_eq!(p.default_permission, AclPermission::Deny);
        // Every policy names rules that exist and a subject that exists.
        for pol in &p.policies {
            for r in &pol.rules {
                assert!(p.rules.iter().any(|x| &x.id == r), "{r} in {}", pol.id);
            }
            for s in &pol.subjects {
                assert!(p.subjects.iter().any(|x| &x.id == s), "{s} in {}", pol.id);
            }
        }
        // Rule ids are unique — zenoh refuses a duplicate.
        let ids: BTreeSet<&str> = p.rules.iter().map(|r| r.id.as_str()).collect();
        assert_eq!(ids.len(), p.rules.len());
    }

    /// Fact 1: one rule per plane, and the host's `**` rule reaches none of
    /// them.
    #[test]
    fn a_host_gets_one_rule_per_plane_and_the_data_rule_reaches_no_plane() {
        let p = plan();
        let pol = policy_of(&p, "h-3fa9c2d41b7e");
        assert_eq!(
            pol.rules,
            [
                "host-data-h-3fa9c2d41b7e",
                "host-serve-h-3fa9c2d41b7e",
                "host-media-h-3fa9c2d41b7e",
                "host-blob-seed-h-3fa9c2d41b7e",
                "host-adv-h-3fa9c2d41b7e",
                INTEREST_PROP,
            ]
        );
        let data = rule_of(&p, "host-data-h-3fa9c2d41b7e");
        assert_eq!(data.key_exprs, ["zensight/v1/h-3fa9c2d41b7e/**"]);
        assert_eq!(data.flows.as_deref(), Some(&[AclFlow::Ingress][..]));
        let data_ke = keyexpr::new(data.key_exprs[0].as_str()).unwrap();
        for plane in [
            "zensight/v1/h-3fa9c2d41b7e/@rpc/sysinfo/introspect",
            "zensight/v1/h-3fa9c2d41b7e/@media/parallax/cam0/video/h264/high",
            "zensight/v1/h-3fa9c2d41b7e/@blob/store/sha256/abc",
            "zensight/v1/h-3fa9c2d41b7e/state/x/@adv/pub/zid/1",
        ] {
            assert!(
                !data_ke.includes(keyexpr::new(plane).unwrap()),
                "host-data must not reach {plane}"
            );
        }
        let serve = rule_of(&p, "host-serve-h-3fa9c2d41b7e");
        assert_eq!(
            serve.flows, None,
            "both directions: query egress, reply ingress"
        );
        assert_eq!(
            serve.key_exprs,
            [
                "zensight/v1/h-3fa9c2d41b7e/@rpc/**",
                "zensight/v1/h-3fa9c2d41b7e/@blob/**"
            ]
        );
        // A plain host: data, serve, interest-prop and nothing else.
        assert_eq!(
            policy_of(&p, "edge-02.example").rules,
            [
                "host-data-h-20609002f7b6",
                "host-serve-h-20609002f7b6",
                INTEREST_PROP
            ]
        );
    }

    /// RFC 06 §1's test vector: the origin is computed from the machine-id,
    /// and a given origin that disagrees is a refusal.
    #[test]
    fn a_machine_id_computes_the_origin_and_a_disagreement_is_refused() {
        let p = plan();
        let s = p
            .subjects
            .iter()
            .find(|s| s.id == "edge-02.example")
            .unwrap();
        assert_eq!(s.cert_common_names, ["edge-02.example"]);
        assert!(rule_of(&p, "host-data-h-20609002f7b6").key_exprs[0].contains("h-20609002f7b6"));

        let mut e = fleet();
        e.principal[1].origin = Some("h-000000000000".into());
        let p = plan_acl(&e, "zensight", None, AclOptions::default());
        assert_eq!(p.refusals.len(), 1);
        assert_eq!(p.refusals[0].principal, "edge-02.example");
        assert!(
            p.refusals[0].reason.contains("h-20609002f7b6"),
            "{}",
            p.refusals[0].reason
        );
        assert_eq!(p.subjects.len(), 5, "the refused principal is left out");

        // No salt: the derivation cannot run.
        let mut e = fleet();
        e.fleet.salt = None;
        let p = plan_acl(&e, "zensight", None, AclOptions::default());
        assert!(p.refusals[0].reason.contains("salt"));
    }

    /// Fact 2: catalog access is its own rule, spelled on the verbatim
    /// origin, and `*` never reaches it.
    #[test]
    fn the_catalog_is_its_own_rule_and_a_star_never_reaches_it() {
        let p = plan();
        let own = rule_of(&p, "catalog-own-catalog");
        assert_eq!(
            own.key_exprs,
            [
                "zensight/v1/@catalog/**",
                "zensight/v1/@catalog/@rpc/**",
                "zensight/v1/@catalog/**/@adv/**"
            ]
        );
        let star = keyexpr::new("zensight/v1/*/state/**").unwrap();
        assert!(!star.includes(keyexpr::new("zensight/v1/@catalog/state/pdns/x").unwrap()));
        // And the consumers name it explicitly, adv sibling included.
        let sub = rule_of(&p, "ops-sub");
        assert!(
            sub.key_exprs
                .contains(&"zensight/v1/@catalog/**".to_string())
        );
        assert!(
            sub.key_exprs
                .contains(&"zensight/v1/@catalog/@rpc/**".to_string())
        );
        assert!(
            sub.key_exprs
                .contains(&"zensight/v1/@catalog/**/@adv/**".to_string())
        );
        assert!(
            !keyexpr::new("zensight/v1/**/@adv/**")
                .unwrap()
                .includes(keyexpr::new("zensight/v1/@catalog/state/x/@adv/pub/z/1").unwrap()),
            "the fifth fact's corollary: `**` cannot cross @catalog"
        );
    }

    /// Fact 4: every consumer policy carries ingress declares.
    #[test]
    fn every_consumer_policy_allows_its_declarations() {
        let p = plan();
        for (pid, rid) in [
            ("zensight-catalog", "catalog-intake-declare-catalog"),
            ("zensight-console", "ops-sub"),
            ("zensight-watch", "watch-sub"),
        ] {
            assert!(policy_of(&p, pid).rules.iter().any(|r| r == rid), "{pid}");
            let r = rule_of(&p, rid);
            assert_eq!(r.flows.as_deref(), Some(&[AclFlow::Ingress][..]));
            for m in [
                AclMessage::DeclareSubscriber,
                AclMessage::DeclareLivelinessSubscriber,
                AclMessage::Query,
            ] {
                assert!(r.messages.contains(&m), "{rid} lacks {}", m.as_str());
            }
        }
        // The catalog's intake is split by flow: a flowless rule would let it
        // tombstone any host's keys.
        let recv = rule_of(&p, "catalog-intake-recv-catalog");
        assert_eq!(recv.flows.as_deref(), Some(&[AclFlow::Egress][..]));
        assert!(recv.messages.contains(&AclMessage::Delete));
        // A watch never sees @media or @blob; a console does.
        let watch = rule_of(&p, "watch-sub");
        assert!(
            !watch
                .key_exprs
                .iter()
                .any(|e| e.contains("@media") || e.contains("@blob"))
        );
        let console = rule_of(&p, "ops-sub");
        assert!(console.key_exprs.iter().any(|e| e.contains("@media")));
        assert!(console.key_exprs.iter().any(|e| e.contains("@blob")));
    }

    /// The fifth fact: `interest-prop` on every publishing policy, egress
    /// only, declares and queries only — and it includes the selectors
    /// consumers actually declare.
    #[test]
    fn interest_prop_rides_every_host_policy_egress_only() {
        let p = plan();
        for pid in ["h-3fa9c2d41b7e", "edge-02.example", "h-0123456789ab"] {
            assert!(
                policy_of(&p, pid).rules.iter().any(|r| r == INTEREST_PROP),
                "{pid}"
            );
        }
        assert!(
            policy_of(&p, "zensight-catalog")
                .rules
                .iter()
                .any(|r| r == INTEREST_PROP)
        );
        let r = rule_of(&p, INTEREST_PROP);
        assert_eq!(r.flows.as_deref(), Some(&[AclFlow::Egress][..]));
        assert!(!r.messages.contains(&AclMessage::Put));
        assert!(!r.messages.contains(&AclMessage::Reply));
        assert!(r.messages.contains(&AclMessage::DeclareSubscriber));
        assert!(r.messages.contains(&AclMessage::Query));
        // The console's own selectors are included, `v1/**` and the fan-out
        // RPC shape among them — neither is included by any own-origin rule.
        let own = keyexpr::new("zensight/v1/h-3fa9c2d41b7e/@rpc/**").unwrap();
        let fanout = keyexpr::new("zensight/v1/*/@rpc/netlink/sockets").unwrap();
        assert!(!own.includes(fanout));
        assert!(
            r.key_exprs
                .iter()
                .any(|e| keyexpr::new(e.as_str()).unwrap().includes(fanout))
        );
        let firehose = keyexpr::new("zensight/v1/**").unwrap();
        assert!(
            r.key_exprs
                .iter()
                .any(|e| keyexpr::new(e.as_str()).unwrap().includes(firehose))
        );
    }

    /// Without a registry the deny is the convention's `set` leaf and the
    /// plan says so; with one it is the declared write set.
    #[test]
    fn the_write_set_is_narrowed_by_the_registry_and_unnarrowed_without() {
        let p = plan();
        let deny = rule_of(&p, NO_REMOTE_ACTIONS);
        assert_eq!(deny.permission, AclPermission::Deny);
        assert_eq!(deny.messages, [AclMessage::Query]);
        assert_eq!(deny.key_exprs, ["zensight/v1/*/@rpc/*/**/set"]);
        assert!(
            p.warnings
                .iter()
                .any(|w| w.kind == AclWarningKind::WriteSetNotNarrowed)
        );
        assert!(p.registry.is_not_asked());
        assert!(
            policy_of(&p, "zensight-console")
                .rules
                .iter()
                .any(|r| r == NO_REMOTE_ACTIONS)
        );
        assert!(
            policy_of(&p, "zensight-watch")
                .rules
                .iter()
                .any(|r| r == NO_REMOTE_ACTIONS)
        );

        let dir =
            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../fixture-tests/registry");
        let slices = SliceSet::from_dirs(&[dir]).unwrap();
        let p = plan_acl(&fleet(), "zensight", Some(&slices), AclOptions::default());
        let facts = p.registry.as_option().unwrap();
        assert!(
            facts
                .write_procedures
                .contains(&"systemd/action/set".to_string())
        );
        assert!(facts.media_producers.contains(&"parallax".to_string()));
        assert!(facts.blob_producers.contains(&"logs".to_string()));
        let deny = rule_of(&p, NO_REMOTE_ACTIONS);
        assert!(
            deny.key_exprs
                .contains(&"zensight/v1/*/@rpc/systemd/action/set".to_string())
        );
        assert!(
            deny.key_exprs
                .contains(&"zensight/v1/@catalog/@rpc/catalog/link".to_string())
        );
        assert!(!deny.key_exprs.iter().any(|e| e.ends_with("/**/set")));
        assert!(
            !p.warnings
                .iter()
                .any(|w| w.kind == AclWarningKind::WriteSetNotNarrowed)
        );
        // A console allowed to act drops the deny.
        let mut e = fleet();
        e.principal[4].remote_actions = true;
        let p = plan_acl(&e, "zensight", Some(&slices), AclOptions::default());
        assert!(
            !policy_of(&p, "zensight-console")
                .rules
                .iter()
                .any(|r| r == NO_REMOTE_ACTIONS)
        );
        assert!(
            policy_of(&p, "zensight-watch")
                .rules
                .iter()
                .any(|r| r == NO_REMOTE_ACTIONS)
        );
    }

    /// The registry narrows the planes: a claimed plane no producer declares
    /// is omitted, and said so.
    #[test]
    fn a_plane_the_registry_does_not_declare_is_omitted_with_a_warning() {
        let dir =
            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../fixture-tests/registry");
        let slices = SliceSet::from_dirs(&[dir]).unwrap();
        // A registry with the media producer removed.
        let kept: Vec<_> = slices
            .slices()
            .iter()
            .filter(|s| s.media.is_empty())
            .cloned()
            .collect();
        let narrowed = SliceSet::from_slices(kept);
        let p = plan_acl(&fleet(), "zensight", Some(&narrowed), AclOptions::default());
        assert!(p.rules.iter().all(|r| !r.id.starts_with("host-media-")));
        let w = p
            .warnings
            .iter()
            .find(|w| w.kind == AclWarningKind::PlaneNotDeclared)
            .expect("a plane warning");
        assert_eq!(w.principal.as_deref(), Some("h-3fa9c2d41b7e"));
        assert!(
            !rule_of(&p, "ops-sub")
                .key_exprs
                .iter()
                .any(|e| e.contains("@media"))
        );
    }

    #[test]
    fn refusals_name_their_reason() {
        let mut e = fleet();
        // Two principals sharing a CN.
        e.principal.push(principal("zensight-console", Role::Watch));
        // A host with neither origin nor machine_id.
        e.principal.push(principal("bare-host", Role::Host));
        // A zid subject without the flag.
        e.principal.push(PrincipalSpec {
            zid: Some("38a4829bce9166ee".into()),
            ..PrincipalSpec {
                cn: None,
                role: Role::Watch,
                ..Default::default()
            }
        });
        // A read-only watch asking to act.
        e.principal.push(PrincipalSpec {
            remote_actions: true,
            ..principal("acting-watch", Role::Watch)
        });
        let p = plan_acl(&e, "zensight", None, AclOptions::default());
        let reasons: Vec<(&str, &str)> = p
            .refusals
            .iter()
            .map(|r| (r.principal.as_str(), r.reason.as_str()))
            .collect();
        assert_eq!(reasons.len(), 4, "{reasons:?}");
        assert!(reasons[0].1.contains("already enrolled"));
        assert!(reasons[1].1.contains("`origin` or `machine_id`"));
        assert_eq!(reasons[2].0, "38a4829bce9166ee");
        assert!(reasons[2].1.contains("--allow-zid-subjects"));
        assert!(reasons[3].1.contains("read-only"));
        assert_eq!(p.subjects.len(), 6);

        // With the flag, the zid subject is admitted and warned about.
        let p = plan_acl(
            &e,
            "zensight",
            None,
            AclOptions {
                allow_zid_subjects: true,
            },
        );
        assert_eq!(p.refusals.len(), 3);
        let s = p
            .subjects
            .iter()
            .find(|s| s.id == "38a4829bce9166ee")
            .unwrap();
        assert_eq!(s.zids, ["38a4829bce9166ee"]);
        assert!(s.cert_common_names.is_empty());
        assert!(
            p.warnings
                .iter()
                .any(|w| w.kind == AclWarningKind::ZidSubject)
        );
    }

    /// #486: the output is a fragment for a router config's top-level
    /// object, so pasted into one it must parse — as zenoh itself reads it.
    fn assert_pastes_into_a_router_config(text: &str) {
        let config = format!("{{\n  mode: \"router\",\n{text}}}\n");
        if let Err(e) = zenoh::Config::from_json5(&config) {
            panic!("the pasted fragment does not parse: {e}\n{config}");
        }
    }

    #[test]
    fn the_json5_names_zenohs_fields_and_every_matrix_row() {
        let text = to_json5(&plan());
        assert_pastes_into_a_router_config(&text);
        assert!(text.starts_with("// zenohd access_control block"));
        assert!(text.contains("access_control: {"));
        assert!(text.contains("  enabled: true,"));
        assert!(text.contains("default_permission: \"deny\""));
        for field in ["rules: [", "subjects: [", "policies: ["] {
            assert!(text.contains(field), "{field}");
        }
        assert!(text.contains(
            "{ id: \"host-data-h-3fa9c2d41b7e\", permission: \"allow\", flows: [\"ingress\"],"
        ));
        assert!(text.contains("messages: [\"put\", \"delete\", \"liveliness_token\"],"));
        // A flowless rule has no `flows` key at all.
        assert!(text.contains("{ id: \"host-serve-h-3fa9c2d41b7e\", permission: \"allow\",\n"));
        assert!(text.contains("// interest-prop: the fifth fact"));
        assert!(text.contains("cert_common_names: [\"h-3fa9c2d41b7e\"] },  // host"));
        assert!(text.contains("// ! write_set_not_narrowed"));
        // And it round-trips through a JSON parser once the comments and the
        // bare keys are what a JSON5 reader accepts — checked here through
        // zenoh's own loader in the zenctl corpus; here, that every quoted
        // string is valid JSON.
        assert!(!text.contains("\\u"));
    }

    #[test]
    fn explain_answers_per_direction_with_the_rules_that_decided() {
        let p = plan();
        // A host's own put: allowed on ingress, nothing on egress.
        let x = explain_acl(
            &p,
            "h-3fa9c2d41b7e",
            "zensight/v1/h-3fa9c2d41b7e/state/sysinfo/health",
            AclMessage::Put,
        )
        .unwrap();
        assert_eq!(x.ingress.decision, AclDecision::Allowed);
        assert_eq!(x.ingress.via[0].rule, "host-data-h-3fa9c2d41b7e");
        assert_eq!(x.egress.decision, AclDecision::DeniedByDefault);

        // Its @rpc reply: host-data does not reach it; host-serve does.
        let x = explain_acl(
            &p,
            "h-3fa9c2d41b7e",
            "zensight/v1/h-3fa9c2d41b7e/@rpc/sysinfo/introspect",
            AclMessage::Reply,
        )
        .unwrap();
        assert_eq!(x.ingress.decision, AclDecision::Allowed);
        assert_eq!(x.ingress.via[0].rule, "host-serve-h-3fa9c2d41b7e");

        // Another host's origin: denied by default, and the reason names the
        // convention.
        let x = explain_acl(
            &p,
            "h-3fa9c2d41b7e",
            "zensight/v1/h-0123456789ab/state/sysinfo/health",
            AclMessage::Put,
        )
        .unwrap();
        assert_eq!(x.ingress.decision, AclDecision::DeniedByDefault);
        assert!(x.ingress.via.is_empty());

        // The console calling a write: denied, deny wins over ops-sub.
        let x = explain_acl(
            &p,
            "zensight-console",
            "zensight/v1/h-3fa9c2d41b7e/@rpc/systemd/action/set",
            AclMessage::Query,
        )
        .unwrap();
        assert_eq!(x.ingress.decision, AclDecision::Denied);
        assert_eq!(x.ingress.via[0].rule, NO_REMOTE_ACTIONS);
        assert_eq!(x.ingress.via[1].rule, "ops-sub");
        assert!(x.ingress.reason.contains("deny wins over ops-sub"));

        // A CN resolves to its subject; an unknown principal is unaskable.
        assert!(explain_acl(&p, "zensight-watch", "zensight/v1/**", AclMessage::Put).is_ok());
        let e = explain_acl(&p, "nobody", "zensight/v1/**", AclMessage::Put).unwrap_err();
        assert!(e.is_unaskable());
        let e = explain_acl(&p, "zensight-watch", "zensight//x", AclMessage::Put).unwrap_err();
        assert!(e.is_unaskable());
    }

    /// The plan's own JSON5, parsed back as zenoh would, checks clean; a
    /// hand-edited block does not.
    #[test]
    fn check_finds_what_differs_and_only_that() {
        let p = plan();
        // The observed side built from the plan itself, as a loader would
        // hand it back.
        let mut doc = AclConfigDoc {
            enabled: true,
            default_permission: "deny".into(),
            rules: p
                .rules
                .iter()
                .map(|r| crate::report::AclRuleDoc {
                    id: r.id.clone(),
                    key_exprs: r.key_exprs.clone(),
                    messages: r.messages.iter().map(|m| m.as_str().to_string()).collect(),
                    flows: r
                        .flows
                        .as_ref()
                        .map(|f| f.iter().map(|x| x.as_str().to_string()).collect()),
                    permission: r.permission.as_str().into(),
                })
                .collect(),
            subjects: p
                .subjects
                .iter()
                .map(|s| crate::report::AclSubjectDoc {
                    id: s.id.clone(),
                    cert_common_names: Some(s.cert_common_names.clone()),
                    ..Default::default()
                })
                .collect(),
            policies: p
                .policies
                .iter()
                .map(|pol| crate::report::AclPolicyDoc {
                    id: None,
                    rules: pol.rules.clone(),
                    subjects: pol.subjects.clone(),
                })
                .collect(),
        };
        let c = check_acl(&p, &doc, "router.json5");
        assert!(c.findings.is_empty(), "{:?}", c.findings);
        assert!(matches!(c.judgement, Judgement::NotEstablished { .. }));
        assert_eq!(c.interest_probe, Judgement::NotAsked);

        // Now break it four ways.
        doc.enabled = false;
        doc.rules.retain(|r| r.id != INTEREST_PROP);
        doc.rules[0].flows = None;
        doc.subjects.push(crate::report::AclSubjectDoc {
            id: "stranger".into(),
            cert_common_names: Some(vec!["not-enrolled".into()]),
            interfaces: Some(vec!["eth0".into()]),
            ..Default::default()
        });
        doc.policies[0].rules.pop();
        let c = check_acl(&p, &doc, "router.json5");
        let kinds: Vec<AclFindingKind> = c.findings.iter().map(|f| f.kind).collect();
        assert!(kinds.contains(&AclFindingKind::Disabled));
        assert!(kinds.contains(&AclFindingKind::RuleMissing));
        assert!(kinds.contains(&AclFindingKind::RuleDiffers));
        assert!(kinds.contains(&AclFindingKind::SubjectExtra));
        assert!(kinds.contains(&AclFindingKind::SubjectUnplannedProperty));
        assert!(kinds.contains(&AclFindingKind::UnknownCn));
        assert!(kinds.contains(&AclFindingKind::PolicyMissing));
        assert!(kinds.contains(&AclFindingKind::PolicyExtra));
        assert_eq!(c.judgement, Judgement::Established);
        assert_eq!(crate::judgement_exit_code(&c.judgement), 1);
    }

    #[test]
    fn a_procedure_path_with_variables_becomes_a_pattern() {
        assert_eq!(procedure_pattern("config/{ns}/{if}/set"), "config/*/*/set");
        assert_eq!(procedure_pattern("action/set"), "action/set");
        assert_eq!(procedure_pattern("subs/{path...}"), "subs/**");
    }

    /// A radio driver's registry, as the reference adopter marks it: sixteen
    /// counters that stay home, four an operator on the far side acts on,
    /// state and events that stay home, every procedure home, and one
    /// sensitive write.
    const MODEM_TOML: &str = "[registry]\nversion = \"1.0\"\napp = \"acme\"\nconvention = 1\n\n[producer]\nname = \"modem\"\n\n\
[[subject]]\npath = \"{device}/tx_sdus_total\"\nclass = \"telemetry\"\ntype = \"C\"\ncardinality = 4\nexposure = \"host\"\n\n\
[[subject]]\npath = \"{device}/rx_sdus_total\"\nclass = \"telemetry\"\ntype = \"C\"\ncardinality = 4\nexposure = \"host\"\n\n\
[[subject]]\npath = \"{device}/queue_depth\"\nclass = \"telemetry\"\ntype = \"G\"\ncardinality = 4\nexposure = \"link\"\n\n\
[[subject]]\npath = \"{device}/dropped_total\"\nclass = \"telemetry\"\ntype = \"C\"\ncardinality = 4\nexposure = \"link\"\n\n\
[[subject]]\npath = \"{device}/temperature\"\nclass = \"telemetry\"\ntype = \"G\"\ncardinality = 4\n\n\
[[subject]]\npath = \"link/{device}\"\nclass = \"state\"\ntype = \"L\"\ncardinality = 4\nttl_s = 60\nexposure = \"host\"\n\n\
[[subject]]\npath = \"endpoint/{device}/{endpoint}\"\nclass = \"state\"\ntype = \"E\"\ncardinality = 16\nttl_s = 60\nexposure = \"host\"\n\n\
[[subject]]\npath = \"link_change/{event_id}\"\nclass = \"events\"\ntype = \"V\"\ncardinality = 1000\nrate = \"low\"\nexposure = \"host\"\n\n\
[[procedure]]\npath = \"introspect\"\nkind = \"read\"\nreply = \"RegistrySlice\"\nexposure = \"host\"\n\n\
[[procedure]]\npath = \"config/{device}/power/set\"\nkind = \"write\"\nreply = \"ConfigView\"\ncardinality = 4\nexposure = \"host\"\n\n\
[[procedure]]\npath = \"config/{device}/access/set\"\nkind = \"write\"\nreply = \"ConfigView\"\ncardinality = 4\nexposure = \"host\"\nsensitive = true\n\n\
[[procedure]]\npath = \"config/{device}/persist\"\nkind = \"write\"\nreply = \"Ack\"\ncardinality = 4\nexposure = \"host\"\n";

    fn modem_slices() -> SliceSet {
        SliceSet::from_slices(vec![zenkey::parse_slice(MODEM_TOML).expect("parses")])
    }

    fn face(interval: LinkInterval) -> FaceSpec {
        FaceSpec {
            id: "modem-lane".into(),
            link_protocols: vec!["unixsock-stream".into()],
            interfaces: Vec::new(),
            interval,
        }
    }

    fn includes(rule: &AclRule, key: &str) -> bool {
        let k = keyexpr::new(key).unwrap();
        rule.key_exprs
            .iter()
            .any(|e| keyexpr::new(e.as_str()).unwrap().includes(k))
    }

    /// A `writes` grant is one allow and a carved deny: the declared writes
    /// the grant does not include, and nothing else. A sensitive write stays
    /// denied to a console that may act, until a grant names it.
    #[test]
    fn a_writes_grant_carves_the_deny_and_a_sensitive_write_stays_denied() {
        let slices = modem_slices();
        let mut e = fleet();
        e.principal[4].writes = vec!["modem/config/*/power/set".into()];
        let p = plan_acl(&e, "", Some(&slices), AclOptions::default());
        assert!(p.refusals.is_empty(), "{:?}", p.refusals);
        let grant = rule_of(&p, "writes-zensight-console");
        assert_eq!(grant.permission, AclPermission::Allow);
        assert_eq!(grant.key_exprs, ["v1/*/@rpc/modem/config/*/power/set"]);
        let deny = rule_of(&p, "no-remote-actions-zensight-console");
        assert!(!includes(
            deny,
            "v1/h-0123456789ab/@rpc/modem/config/rf0/power/set"
        ));
        assert!(includes(
            deny,
            "v1/h-0123456789ab/@rpc/modem/config/rf0/persist"
        ));
        assert!(includes(
            deny,
            "v1/h-0123456789ab/@rpc/modem/config/rf0/access/set"
        ));
        // The watch keeps the shared, whole deny.
        assert!(
            policy_of(&p, "zensight-watch")
                .rules
                .iter()
                .any(|r| r == NO_REMOTE_ACTIONS)
        );
        assert!(
            p.registry
                .as_option()
                .unwrap()
                .sensitive_procedures
                .contains(&"modem/config/{device}/access/set".to_string())
        );
        // `--explain` answers the per-resource question.
        let x = explain_acl(
            &p,
            "zensight-console",
            "v1/h-0123456789ab/@rpc/modem/config/rf0/power/set",
            AclMessage::Query,
        )
        .unwrap();
        assert_eq!(x.ingress.decision, AclDecision::Allowed, "{x:?}");

        // A console that may act still cannot reach the sensitive write —
        // until a grant names it.
        let mut e = fleet();
        e.principal[4].remote_actions = true;
        let p = plan_acl(&e, "", Some(&slices), AclOptions::default());
        let deny = rule_of(&p, "no-remote-actions-zensight-console");
        assert_eq!(deny.key_exprs, ["v1/*/@rpc/modem/config/*/access/set"]);
        e.principal[4].writes = vec!["modem/config/*/access/set".into()];
        let p = plan_acl(&e, "", Some(&slices), AclOptions::default());
        assert!(
            !policy_of(&p, "zensight-console")
                .rules
                .iter()
                .any(|r| r.starts_with("no-remote-actions")),
            "nothing left to deny the console; the watch keeps its own"
        );

        // Without a registry a grant cannot carve: the deny stays whole and
        // the plan says why.
        let mut e = fleet();
        e.principal[4].writes = vec!["modem/config/*/power/set".into()];
        let p = plan_acl(&e, "", None, AclOptions::default());
        assert!(!p.rules.iter().any(|r| r.id.starts_with("writes-")));
        assert!(rule_of(&p, NO_REMOTE_ACTIONS).key_exprs[0].ends_with("/**/set"));
        assert!(
            p.warnings
                .iter()
                .any(|w| w.kind == AclWarningKind::GrantNotNarrowed)
        );
        // A pattern nothing answers is allowed as spelled, and said.
        let mut e = fleet();
        e.principal[4].writes = vec!["modem/reboot".into()];
        let p = plan_acl(&e, "", Some(&slices), AclOptions::default());
        assert!(
            p.warnings
                .iter()
                .any(|w| w.kind == AclWarningKind::GrantMatchesNothing)
        );
        // A watch refuses a grant as it refuses remote_actions.
        let mut e = fleet();
        e.principal[5].writes = vec!["modem/config/*/power/set".into()];
        let p = plan_acl(&e, "", Some(&slices), AclOptions::default());
        assert_eq!(p.refusals.len(), 1);
        assert!(p.refusals[0].reason.contains("read-only"));
    }

    /// The radio face (RFC 09 §4): the `host` counters denied by name, the
    /// wholly-`host` classes by subtree, the planes by their widest
    /// spelling, the `link` counters capped — and the unmarked one crosses,
    /// said.
    #[test]
    fn a_constrained_face_keeps_host_surfaces_home_and_caps_link_ones() {
        let slices = modem_slices();
        let p = plan_face(&slices, "", &face(LinkInterval::EverySecs(60)));
        assert_eq!(p.default_permission, AclPermission::Allow);
        assert_eq!(p.subjects.len(), 1);
        assert_eq!(p.subjects[0].link_protocols, ["unixsock-stream"]);
        assert_eq!(p.subjects[0].role, Role::Link);
        assert_eq!(p.policies[0].subjects, ["modem-lane"]);
        assert_eq!(p.policies[0].rules.len(), p.rules.len());
        assert!(p.rules.iter().all(|r| r.flows.as_deref() == Some(BOTH)));

        let detail = rule_of(&p, "deny-telemetry-modem-detail");
        assert!(includes(
            detail,
            "v1/h-0123456789ab/telemetry/modem/rf0/tx_sdus_total"
        ));
        assert!(!includes(
            detail,
            "v1/h-0123456789ab/telemetry/modem/rf0/queue_depth"
        ));
        assert!(!includes(
            detail,
            "v1/h-0123456789ab/telemetry/modem/rf0/temperature"
        ));
        let state = rule_of(&p, "deny-state-modem");
        assert_eq!(state.key_exprs, ["v1/*/state/modem/**"]);
        assert!(includes(state, "v1/h-0123456789ab/state/modem/alive"));
        assert_eq!(state.messages.len(), AclMessage::ALL.len());
        assert_eq!(
            rule_of(&p, "deny-events-modem").key_exprs,
            ["v1/*/events/modem/**"]
        );
        let rpc = rule_of(&p, "deny-rpc");
        assert_eq!(rpc.key_exprs, ["**/@rpc/**"]);
        assert!(
            includes(rpc, "v1/*/@rpc/**"),
            "the broader query is included too"
        );
        assert!(includes(rule_of(&p, "deny-adminspace"), "@/abc/router/**"));
        assert!(includes(
            rule_of(&p, "deny-adminspace"),
            "v1/h-0123456789ab/state/modem/link/rf0/@adv/x"
        ));
        assert!(includes(
            rule_of(&p, "deny-media"),
            "v1/h-0123456789ab/@media/cam/main"
        ));

        let capped: Vec<&str> = p.downsampling.iter().map(|d| d.key_expr.as_str()).collect();
        assert_eq!(
            capped,
            [
                "v1/*/telemetry/modem/*/dropped_total",
                "v1/*/telemetry/modem/*/queue_depth"
            ]
        );
        assert!(p.downsampling.iter().all(|d| d.freq <= 0.017));
        let crosses = p
            .warnings
            .iter()
            .find(|w| w.kind == AclWarningKind::FaceCrosses)
            .expect("the unmarked gauge is said");
        assert!(
            crosses.text.contains("telemetry/{device}/temperature"),
            "{}",
            crosses.text
        );
        assert!(
            !p.warnings
                .iter()
                .any(|w| w.kind == AclWarningKind::FaceCutsPlane)
        );

        // The block, as zenohd reads it: the face's subject, both blocks.
        let text = to_json5(&p);
        assert_pastes_into_a_router_config(&text);
        assert!(text.contains("default_permission: \"allow\""));
        assert!(text.contains("link_protocols: [\"unixsock-stream\"] },  // link"));
        assert!(text.contains("downsampling: ["));
        assert!(text.contains("flows: [\"egress\"], messages: [\"put\"]"));
        // And --explain over the face: the deny decided, on ingress and egress.
        let x = explain_acl(
            &p,
            "modem-lane",
            "v1/h-0123456789ab/telemetry/modem/rf0/tx_sdus_total",
            AclMessage::Put,
        )
        .unwrap();
        assert_eq!(x.egress.decision, AclDecision::Denied);
        let x = explain_acl(
            &p,
            "modem-lane",
            "v1/h-0123456789ab/telemetry/modem/rf0/queue_depth",
            AclMessage::Put,
        )
        .unwrap();
        assert_eq!(
            x.egress.decision,
            AclDecision::AllowedByDefault,
            "capped, not denied"
        );

        // No affordable rate (a billed channel): the `link` counters stay
        // home too, the class is denied whole, nothing is capped.
        let p = plan_face(&slices, "", &face(LinkInterval::None));
        assert!(p.downsampling.is_empty());
        let detail = rule_of(&p, "deny-telemetry-modem-detail");
        assert!(includes(
            detail,
            "v1/h-0123456789ab/telemetry/modem/rf0/queue_depth"
        ));
        assert!(!includes(
            detail,
            "v1/h-0123456789ab/telemetry/modem/rf0/temperature"
        ));
        assert!(!to_json5(&p).contains("downsampling"));

        // A registry that marks nothing crosses whole, and the note says so;
        // a `link` procedure beside a `host` one is cut with the plane.
        let dir =
            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../fixture-tests/registry");
        let fixtures = SliceSet::from_dirs(&[dir]).unwrap();
        let p = plan_face(&fixtures, "zensight", &face(LinkInterval::EverySecs(10)));
        assert!(
            p.rules.iter().any(|r| r.id == "deny-rpc"),
            "systemd marks a write host"
        );
        assert!(
            p.warnings
                .iter()
                .any(|w| w.kind == AclWarningKind::FaceCutsPlane)
        );
        assert!(
            p.warnings
                .iter()
                .any(|w| w.kind == AclWarningKind::FaceCrosses)
        );
        assert_eq!(p.downsampling.len(), 1, "sysinfo memory/used is link");
        assert!(
            p.downsampling[0]
                .key_expr
                .starts_with("zensight/v1/*/telemetry/sysinfo/memory/used")
        );
    }
}