epics-bridge-rs 0.18.2

//! Tower-style middleware for the PVA gateway.
//!
//! Wraps the underlying [`epics_pva_rs::server_native::ChannelSource`]
//! ([`super::source::GatewayChannelSource`] in practice) with
//! composable [`Layer`]s that add cross-cutting concerns:
//!
//! - [`AclLayer`] — refuse `has_pv` / `put_value` for PV names that
//!   match the deny list. Patterns may be glob (`*` wildcard) or,
//!   via [`AclConfig::deny_regex`] / [`AclConfig::allow_regex`],
//!   full anchored regular expressions (B7).
//! - [`ReadOnlyLayer`] — fail every PUT, even if upstream allows
//!   it. Mirrors the existing `read_only` flag but as a composable
//!   layer so operators can stack it with audit / ACL.
//! - [`AuditLayer`] — emit a structured event for every PUT,
//!   reusing the existing audit pipeline shape
//!
//! Design constraint: the inner [`ChannelSource`] is the gateway's
//! own [`super::source::GatewayChannelSource`], and we want the
//! `Layer` chain to short-circuit BEFORE the call reaches it (so an
//! ACL deny doesn't trigger an upstream search). Each [`Layer`]
//! implementation forwards calls verbatim by default; override the
//! method to insert pre/post hooks.

use std::sync::Arc;

use epics_pva_rs::pvdata::{FieldDesc, PvField};
use epics_pva_rs::server_native::ChannelContext;
use epics_pva_rs::server_native::source::{ChannelSource, RawMonitorEvent};
use tokio::sync::mpsc;

/// Wrap a [`ChannelSource`] and produce a new one with extra
/// behaviour. Implementations override only the methods they need;
/// the default forwards every call unchanged.
pub trait Layer<S: ChannelSource>: Send + Sync + 'static {
    type Wrapped: ChannelSource;
    fn layer(self, inner: S) -> Self::Wrapped;
}

// ── ReadOnlyLayer ────────────────────────────────────────────────

/// Reject every PUT regardless of upstream policy. Composable
/// with audit / ACL — stacking `ReadOnlyLayer` last in the chain
/// guarantees no PUT can reach the underlying source even if a
/// later layer would have allowed it.
pub struct ReadOnlyLayer;

pub struct ReadOnly<S> {
    inner: Arc<S>,
}

impl<S: ChannelSource> Layer<S> for ReadOnlyLayer {
    type Wrapped = ReadOnly<S>;
    fn layer(self, inner: S) -> ReadOnly<S> {
        ReadOnly {
            inner: Arc::new(inner),
        }
    }
}

impl<S: ChannelSource> ChannelSource for ReadOnly<S> {
    // Round 43: forward the inner source's AccessGate so the wire
    // layer's `gate.check` flows to the actual policy holder, not
    // a permissive Open singleton.
    fn access(&self) -> &epics_pva_rs::server_native::source::AccessGate {
        self.inner.access()
    }
    async fn list_pvs(&self) -> Vec<String> {
        self.inner.list_pvs().await
    }
    async fn has_pv(&self, name: &str) -> bool {
        self.inner.has_pv(name).await
    }
    async fn get_introspection(&self, name: &str) -> Option<FieldDesc> {
        self.inner.get_introspection(name).await
    }
    async fn get_value(&self, name: &str) -> Option<PvField> {
        self.inner.get_value(name).await
    }
    async fn put_value(&self, _name: &str, _value: PvField) -> Result<(), String> {
        Err("read-only mode: PUT rejected".into())
    }
    async fn put_value_checked(
        &self,
        _checked: epics_pva_rs::server_native::source::AccessChecked,
        _value: PvField,
        _ctx: ChannelContext,
    ) -> Result<(), String> {
        Err("read-only mode: PUT rejected".into())
    }
    // A BitSet-delta PUT is still a PUT. Reject it here exactly the
    // way `put_value_checked` does — without this override the trait
    // default would run get_value + merge + put_value_checked, which
    // happens to still reject (put_value_checked above), but only
    // after a wasted read-merge and after bypassing the inner
    // source's atomic `put_delta_checked`. Short-circuit instead.
    async fn put_delta_checked(
        &self,
        _checked: epics_pva_rs::server_native::source::AccessChecked,
        _desc: FieldDesc,
        _changed: epics_pva_rs::proto::BitSet,
        _delta: PvField,
        _ctx: ChannelContext,
    ) -> Result<(), String> {
        Err("read-only mode: PUT rejected".into())
    }
    // PROCESS mutates upstream record state — it is a WRITE-class op,
    // so reject it here exactly the way `put_value` does. Without this
    // override the trait-default `process` (`Ok(())`) would falsely
    // report success and let a PROCESS slip past read-only mode.
    async fn process(&self, _name: &str) -> Result<(), String> {
        Err("read-only mode: PROCESS rejected".into())
    }
    // Typed PROCESS — same WRITE-class rejection as `process` above.
    // Without this override the trait-default `process_checked` would
    // fall through to `process` (rejected) but only after the gate
    // check; short-circuit here so a PROCESS never reaches the inner
    // source under read-only mode.
    async fn process_checked(
        &self,
        _checked: epics_pva_rs::server_native::source::AccessChecked,
        _ctx: ChannelContext,
    ) -> Result<(), String> {
        Err("read-only mode: PROCESS rejected".into())
    }
    async fn is_writable(&self, _name: &str) -> bool {
        false
    }
    async fn subscribe(&self, name: &str) -> Option<mpsc::Receiver<PvField>> {
        self.inner.subscribe(name).await
    }
    // Forward type-state op variants to the inner. Without these,
    // the trait defaults would route through this layer's ctx-less
    // (`get_value`/`subscribe`/`subscribe_raw`/`rpc`) — silently
    // skipping any ctx-aware behaviour the inner installs (gateway
    // per-credential routing, audit ctx fields).
    async fn get_value_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Option<PvField> {
        self.inner.get_value_checked(checked, ctx).await
    }
    async fn subscribe_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Option<mpsc::Receiver<PvField>> {
        self.inner.subscribe_checked(checked, ctx).await
    }
    async fn subscribe_raw(&self, name: &str) -> Option<mpsc::Receiver<RawMonitorEvent>> {
        self.inner.subscribe_raw(name).await
    }
    async fn subscribe_raw_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Option<mpsc::Receiver<RawMonitorEvent>> {
        self.inner.subscribe_raw_checked(checked, ctx).await
    }
    // BR-R14: forward the event-affecting-options MONITOR variants to
    // the inner. Without these the trait default drops `opts` and
    // routes through `subscribe_checked`, so the gateway source never
    // sees the options it must reject. A read-only gateway places no
    // extra constraint on a (read) MONITOR — pure pass-through.
    async fn subscribe_checked_opts(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
        opts: epics_pva_rs::server_native::MonitorOptions,
    ) -> Option<mpsc::Receiver<PvField>> {
        self.inner.subscribe_checked_opts(checked, ctx, opts).await
    }
    async fn subscribe_raw_checked_opts(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
        opts: epics_pva_rs::server_native::MonitorOptions,
    ) -> Option<mpsc::Receiver<RawMonitorEvent>> {
        self.inner
            .subscribe_raw_checked_opts(checked, ctx, opts)
            .await
    }
    async fn rpc(
        &self,
        name: &str,
        request_desc: FieldDesc,
        request_value: PvField,
    ) -> Result<(FieldDesc, PvField), String> {
        self.inner.rpc(name, request_desc, request_value).await
    }
    async fn rpc_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        request_desc: FieldDesc,
        request_value: PvField,
        ctx: ChannelContext,
    ) -> Result<(FieldDesc, PvField), String> {
        self.inner
            .rpc_checked(checked, request_desc, request_value, ctx)
            .await
    }
    fn notify_watermark_high(&self, name: &str) {
        self.inner.notify_watermark_high(name);
    }
    fn notify_watermark_low(&self, name: &str) {
        self.inner.notify_watermark_low(name);
    }
}

// ── AclLayer ─────────────────────────────────────────────────────

/// Pattern-matched access control. PV names matching any `deny`
/// pattern are rejected at the layer before reaching the upstream
/// proxy. `allow_only` (when non-empty) flips the policy — only
/// names matching one of these patterns get through.
///
/// B7: two pattern syntaxes are supported and may be mixed freely.
///
/// * **Glob** — the `deny` / `allow_only` `Vec<String>` fields.
///   `*` matches any run of characters; a pattern with no `*` is an
///   exact match. Backward-compatible with the original AclConfig.
/// * **Regex** — added via [`AclConfig::deny_regex`] /
///   [`AclConfig::allow_regex`]. The pattern is a full
///   [`regex`]-crate regular expression, **anchored** at both ends
///   (the matcher wraps it as `^(?:pattern)$`) so `BL10C:.*`
///   matches the same names a `BL10C:*` glob would, and a bare
///   `MOTOR:VAL` regex matches only that exact name. Regexes are
///   compiled once at config-build time, not per PV check.
///
/// A name is allowed iff it matches **no** deny pattern (glob or
/// regex) AND, when any allow pattern is configured, matches **at
/// least one** allow pattern (glob or regex). Deny always wins.
#[derive(Clone, Default)]
pub struct AclConfig {
    /// Glob deny patterns (`*` wildcard / exact match).
    pub deny: Vec<String>,
    /// Glob allow-only patterns (`*` wildcard / exact match).
    pub allow_only: Vec<String>,
    /// B7: compiled regex deny patterns. Built via
    /// [`AclConfig::deny_regex`]; anchored at both ends.
    deny_re: Vec<regex::Regex>,
    /// B7: compiled regex allow-only patterns. Built via
    /// [`AclConfig::allow_regex`]; anchored at both ends.
    allow_re: Vec<regex::Regex>,
}

impl AclConfig {
    /// B7: add a regex pattern to the deny list. The pattern is
    /// anchored (`^(?:pattern)$`) and compiled immediately; an
    /// invalid pattern returns the [`regex::Error`] so the operator
    /// learns about the typo at config time, not on the first PV
    /// check.
    pub fn deny_regex(mut self, pattern: &str) -> Result<Self, regex::Error> {
        self.deny_re.push(compile_anchored(pattern)?);
        Ok(self)
    }

    /// B7: add a regex pattern to the allow-only list. Anchored and
    /// compiled like [`Self::deny_regex`]. As with glob
    /// `allow_only`, a non-empty allow list flips the policy to
    /// default-deny.
    pub fn allow_regex(mut self, pattern: &str) -> Result<Self, regex::Error> {
        self.allow_re.push(compile_anchored(pattern)?);
        Ok(self)
    }

    /// True iff any allow pattern (glob or regex) is configured —
    /// i.e. the policy is default-deny.
    fn has_allow_list(&self) -> bool {
        !self.allow_only.is_empty() || !self.allow_re.is_empty()
    }

    pub fn allowed(&self, name: &str) -> bool {
        // Deny always wins — glob or regex.
        if self.deny.iter().any(|p| matches_pattern(p, name))
            || self.deny_re.iter().any(|re| re.is_match(name))
        {
            return false;
        }
        // Allow-only: when configured, the name must match at least
        // one allow pattern (glob or regex).
        if self.has_allow_list() {
            let allowed = self.allow_only.iter().any(|p| matches_pattern(p, name))
                || self.allow_re.iter().any(|re| re.is_match(name));
            if !allowed {
                return false;
            }
        }
        true
    }
}

/// B7: per-pattern compiled-program size limit for operator-supplied
/// ACL regexes. 256 KiB is far more than any realistic PV-name
/// pattern needs (a generous allow/deny list compiles to a few KiB)
/// while still rejecting a pathological pattern up front instead of
/// letting it consume unbounded memory at compile time.
const ACL_REGEX_SIZE_LIMIT: usize = 256 * 1024;

/// B7: compile `pattern` as a both-ends-anchored regex so a regex
/// ACL entry matches whole PV names, mirroring glob semantics
/// (`BL10C:*` ≡ `BL10C:.*`, bare token ≡ exact match). Wrapping in a
/// non-capturing group keeps top-level alternation (`a|b`) anchored
/// as `^(?:a|b)$` rather than `^a|b$`.
///
/// Compiled via [`regex::RegexBuilder`] with explicit `size_limit`
/// and `dfa_size_limit` so an operator-supplied pathological pattern
/// fails fast with a bounded `regex::Error` rather than compiling
/// with the crate defaults (which allow far larger programs).
fn compile_anchored(pattern: &str) -> Result<regex::Regex, regex::Error> {
    regex::RegexBuilder::new(&format!("^(?:{pattern})$"))
        .size_limit(ACL_REGEX_SIZE_LIMIT)
        .dfa_size_limit(ACL_REGEX_SIZE_LIMIT)
        .build()
}

fn matches_pattern(pattern: &str, name: &str) -> bool {
    if let Some(prefix) = pattern.strip_suffix('*') {
        return name.starts_with(prefix);
    }
    if let Some(suffix) = pattern.strip_prefix('*') {
        return name.ends_with(suffix);
    }
    name == pattern
}

pub struct AclLayer {
    config: AclConfig,
}

impl AclLayer {
    pub fn new(config: AclConfig) -> Self {
        Self { config }
    }
}

pub struct Acl<S> {
    inner: Arc<S>,
    config: AclConfig,
}

impl<S: ChannelSource> Layer<S> for AclLayer {
    type Wrapped = Acl<S>;
    fn layer(self, inner: S) -> Acl<S> {
        Acl {
            inner: Arc::new(inner),
            config: self.config,
        }
    }
}

impl<S: ChannelSource> ChannelSource for Acl<S> {
    async fn list_pvs(&self) -> Vec<String> {
        // Filter the underlying list so introspection sweeps don't
        // leak the names of denied PVs.
        let mut names = self.inner.list_pvs().await;
        names.retain(|n| self.config.allowed(n));
        names
    }
    async fn has_pv(&self, name: &str) -> bool {
        if !self.config.allowed(name) {
            return false;
        }
        self.inner.has_pv(name).await
    }
    async fn get_introspection(&self, name: &str) -> Option<FieldDesc> {
        if !self.config.allowed(name) {
            return None;
        }
        self.inner.get_introspection(name).await
    }
    async fn get_value(&self, name: &str) -> Option<PvField> {
        if !self.config.allowed(name) {
            return None;
        }
        self.inner.get_value(name).await
    }
    async fn put_value(&self, name: &str, value: PvField) -> Result<(), String> {
        if !self.config.allowed(name) {
            return Err(format!("ACL: PV '{name}' denied"));
        }
        self.inner.put_value(name, value).await
    }
    // PROCESS mutates upstream record state — a WRITE-class op. Gate
    // it by the layer's static allowlist BEFORE forwarding, exactly
    // the way `put_value` is gated. Without this override the trait
    // default `process` (`Ok(())`) bypasses the ACL entirely.
    async fn process(&self, name: &str) -> Result<(), String> {
        if !self.config.allowed(name) {
            return Err(format!("ACL: PV '{name}' denied"));
        }
        self.inner.process(name).await
    }
    // Round 43: type-state op variants gate by the layer's static
    // allowlist BEFORE delegating. The inner source still gets the
    // full AccessChecked + ctx and may apply its own ACF / per-
    // credential routing on top.
    fn access(&self) -> &epics_pva_rs::server_native::source::AccessGate {
        self.inner.access()
    }
    async fn get_value_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Option<PvField> {
        if !self.config.allowed(checked.pv_name()) {
            return None;
        }
        self.inner.get_value_checked(checked, ctx).await
    }
    async fn put_value_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        value: PvField,
        ctx: ChannelContext,
    ) -> Result<(), String> {
        if !self.config.allowed(checked.pv_name()) {
            return Err(format!("ACL: PV '{}' denied", checked.pv_name()));
        }
        self.inner.put_value_checked(checked, value, ctx).await
    }
    // A BitSet-delta PUT is a PUT — gate it by the same static
    // allowlist as `put_value_checked`, then forward to the inner's
    // `put_delta_checked`. Without this override the trait default
    // would run get_value + merge + put_value_checked on THIS layer,
    // bypassing the inner source's atomic `put_delta_checked`
    // (`SharedSource` / `CompositeSource` merge under one lock) and
    // re-opening the concurrent-partial-PUT lost-update window.
    async fn put_delta_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        desc: FieldDesc,
        changed: epics_pva_rs::proto::BitSet,
        delta: PvField,
        ctx: ChannelContext,
    ) -> Result<(), String> {
        if !self.config.allowed(checked.pv_name()) {
            return Err(format!("ACL: PV '{}' denied", checked.pv_name()));
        }
        self.inner
            .put_delta_checked(checked, desc, changed, delta, ctx)
            .await
    }
    async fn is_writable(&self, name: &str) -> bool {
        self.config.allowed(name) && self.inner.is_writable(name).await
    }
    async fn subscribe(&self, name: &str) -> Option<mpsc::Receiver<PvField>> {
        if !self.config.allowed(name) {
            return None;
        }
        self.inner.subscribe(name).await
    }
    async fn subscribe_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Option<mpsc::Receiver<PvField>> {
        if !self.config.allowed(checked.pv_name()) {
            return None;
        }
        self.inner.subscribe_checked(checked, ctx).await
    }
    async fn subscribe_raw(&self, name: &str) -> Option<mpsc::Receiver<RawMonitorEvent>> {
        if !self.config.allowed(name) {
            return None;
        }
        self.inner.subscribe_raw(name).await
    }
    async fn subscribe_raw_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Option<mpsc::Receiver<RawMonitorEvent>> {
        if !self.config.allowed(checked.pv_name()) {
            return None;
        }
        self.inner.subscribe_raw_checked(checked, ctx).await
    }
    // BR-R14: forward the event-affecting-options MONITOR variants,
    // applying the same ACL deny-list gate as `subscribe_*_checked`.
    async fn subscribe_checked_opts(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
        opts: epics_pva_rs::server_native::MonitorOptions,
    ) -> Option<mpsc::Receiver<PvField>> {
        if !self.config.allowed(checked.pv_name()) {
            return None;
        }
        self.inner.subscribe_checked_opts(checked, ctx, opts).await
    }
    async fn subscribe_raw_checked_opts(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
        opts: epics_pva_rs::server_native::MonitorOptions,
    ) -> Option<mpsc::Receiver<RawMonitorEvent>> {
        if !self.config.allowed(checked.pv_name()) {
            return None;
        }
        self.inner
            .subscribe_raw_checked_opts(checked, ctx, opts)
            .await
    }
    async fn rpc(
        &self,
        name: &str,
        request_desc: FieldDesc,
        request_value: PvField,
    ) -> Result<(FieldDesc, PvField), String> {
        if !self.config.allowed(name) {
            return Err(format!("ACL: PV '{name}' denied"));
        }
        self.inner.rpc(name, request_desc, request_value).await
    }
    async fn rpc_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        request_desc: FieldDesc,
        request_value: PvField,
        ctx: ChannelContext,
    ) -> Result<(FieldDesc, PvField), String> {
        if !self.config.allowed(checked.pv_name()) {
            return Err(format!("ACL: PV '{}' denied", checked.pv_name()));
        }
        self.inner
            .rpc_checked(checked, request_desc, request_value, ctx)
            .await
    }
    // Typed PROCESS — gate by the static allowlist BEFORE forwarding,
    // mirroring `rpc_checked` / `put_value_checked`. The inner source
    // still gets the full AccessChecked + ctx and may apply its own
    // ACF gate (PROCESS is WRITE-class) on top.
    async fn process_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Result<(), String> {
        if !self.config.allowed(checked.pv_name()) {
            return Err(format!("ACL: PV '{}' denied", checked.pv_name()));
        }
        self.inner.process_checked(checked, ctx).await
    }
    fn notify_watermark_high(&self, name: &str) {
        self.inner.notify_watermark_high(name);
    }
    fn notify_watermark_low(&self, name: &str) {
        self.inner.notify_watermark_low(name);
    }
}

// ── AuditLayer ───────────────────────────────────────────────────

/// Hook fired on every PUT. Ops typically wire this into a file
/// sink (line-based JSON, libca-asLib text, etc.) — the layer
/// stays format-agnostic.
///
/// **Implementation contract**: `record` is called synchronously
/// from inside the `put_value` async path. Any blocking I/O the
/// implementation does (file write, network send) blocks the
/// calling tokio worker thread for the duration of the PUT, so
/// real-world implementations should either:
/// - keep `record` purely in-memory (counter increment, mpsc
///   try_send into a background drain task), or
/// - use the bundled mpsc-buffered wrapper (see
///   `epics_ca_rs::audit::AuditLogger` for the same pattern).
///
/// The default `ClosureAudit` is fine for tests and counters.
pub trait AuditSink: Send + Sync + 'static {
    fn record(&self, event: AuditEvent);
}

/// Default no-op sink — useful when wiring a layer chain in tests.
pub struct NoopAudit;

impl AuditSink for NoopAudit {
    fn record(&self, _event: AuditEvent) {}
}

/// Forward through a shared / type-erased sink. Lets a config carry an
/// `Arc<dyn AuditSink>` (the gateway's audit-sink plumbing) and still
/// satisfy `AuditLayer::new`'s concrete `A: AuditSink` bound. Covers
/// both `Arc<ConcreteSink>` and `Arc<dyn AuditSink>`.
impl<A: AuditSink + ?Sized> AuditSink for Arc<A> {
    fn record(&self, event: AuditEvent) {
        (**self).record(event);
    }
}

/// Boxed-closure audit sink — convenient for inline tests +
/// custom integrations without a dedicated trait impl.
pub struct ClosureAudit<F: Fn(AuditEvent) + Send + Sync + 'static>(pub F);

impl<F: Fn(AuditEvent) + Send + Sync + 'static> AuditSink for ClosureAudit<F> {
    fn record(&self, event: AuditEvent) {
        (self.0)(event);
    }
}

/// Bounded-mpsc adapter that drains audit events on a background
/// task. Use this when the underlying sink does blocking I/O
/// (file write, network send, syslog) — the AuditLayer's
/// `record()` becomes a non-blocking `try_send` that drops on
/// queue overflow rather than stalling the PUT path.
///
/// The drainer task keeps running until both: (a) every clone of
/// this sink has been dropped, and (b) the receiver has drained.
/// Drop order matters in shutdown — drop the gateway / Arc<Audited>
/// chain BEFORE waiting on `.flush()` to avoid leaving events
/// in flight.
///
/// Mirrors the pattern in `epics_ca_rs::audit::AuditLogger` but
/// generalised to the gateway's `AuditEvent` shape.
pub struct MpscAuditSink {
    tx: tokio::sync::mpsc::Sender<AuditEvent>,
    /// Counter of events dropped due to a full queue. Read via
    /// [`Self::drops`] for diagnostics. Drops happen when the
    /// blocking sink can't keep up — losing audit events under
    /// sustained overload is strictly better than pinning a
    /// downstream PUT.
    drops: Arc<std::sync::atomic::AtomicU64>,
}

impl MpscAuditSink {
    /// Wrap a blocking sink (anything that impls AuditSink) in a
    /// bounded queue. `capacity` is the max in-flight events; past
    /// that the layer's `record()` becomes a no-op + drop counter
    /// increment. `inner` runs on the spawned drainer task — its
    /// `record()` is allowed to block.
    pub fn wrap<A: AuditSink>(capacity: usize, inner: A) -> Self {
        let (tx, mut rx) = tokio::sync::mpsc::channel::<AuditEvent>(capacity.max(1));
        tokio::spawn(async move {
            while let Some(ev) = rx.recv().await {
                inner.record(ev);
            }
        });
        Self {
            tx,
            drops: Arc::new(std::sync::atomic::AtomicU64::new(0)),
        }
    }

    /// Number of audit events dropped due to a full queue. Stays
    /// at 0 in normal operation; growing values mean the sink is
    /// slower than the PUT rate and the operator should look at
    /// the underlying I/O stack.
    pub fn drops(&self) -> u64 {
        self.drops.load(std::sync::atomic::Ordering::Relaxed)
    }
}

impl AuditSink for MpscAuditSink {
    fn record(&self, event: AuditEvent) {
        if self.tx.try_send(event).is_err() {
            self.drops
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
    }
}

#[derive(Debug, Clone)]
pub struct AuditEvent {
    /// PV name the operation targeted.
    pub pv: String,
    /// Operation type. Currently only PUT triggers the layer; new
    /// variants will be additive when other op types start
    /// auditing.
    pub event: AuditEventKind,
    /// Authenticated user from the downstream peer's
    /// `ChannelContext`. Empty for `put_value` (non-credentialed)
    /// path.
    pub user: String,
    /// Authenticated host. Same caveat as `user`.
    pub host: String,
    /// Outcome — see [`AuditResult`].
    pub result: AuditResult,
    /// Wall-clock at the moment `record()` was called. Useful for
    /// log shippers that need their own canonical timestamp rather
    /// than the time-of-write.
    pub timestamp: std::time::SystemTime,
    /// Error message body when `result` is `Failed` / `Denied`.
    /// Empty otherwise.
    pub error: String,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuditEventKind {
    /// PUT operation — the layer always audits these.
    Put,
    /// GET operation. Audited only when [`AuditLayer::with_get`]
    /// is enabled; defaults off because GET frequency is
    /// typically much higher than PUT.
    Get,
    /// Subscribe / monitor INIT. Logged when an audit-enabled
    /// layer wraps a source whose `subscribe` returns a fresh
    /// receiver. Distinct from individual update events.
    Subscribe,
    /// RPC dispatch.
    Rpc,
    /// PROCESS operation (PVA wire cmd 16) — record-state mutation
    /// without a value payload. WRITE-class for ACF, always audited
    /// alongside PUT.
    Process,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuditResult {
    Ok,
    Denied,
    Failed,
}

/// Build an [`AuditEvent`] from the inner-call outcome. Detects
/// `Denied` vs `Failed` heuristically: error messages produced by
/// [`AclLayer`] / [`ReadOnlyLayer`] / source-level ACF checks
/// usually contain "denied" / "deny" / "ACL" / "read-only", so the
/// downstream operator gets the right bucket without each layer
/// having to invent a structured-error type.
fn make_audit_event(name: &str, user: &str, host: &str, result: &Result<(), String>) -> AuditEvent {
    let (kind, error) = match result {
        Ok(_) => (AuditResult::Ok, String::new()),
        Err(msg) => {
            let lower = msg.to_lowercase();
            if lower.contains("deny")
                || lower.contains("denied")
                || lower.contains("acl:")
                || lower.contains("read-only")
            {
                (AuditResult::Denied, msg.clone())
            } else {
                (AuditResult::Failed, msg.clone())
            }
        }
    };
    AuditEvent {
        pv: name.to_string(),
        event: AuditEventKind::Put,
        user: user.to_string(),
        host: host.to_string(),
        result: kind,
        timestamp: std::time::SystemTime::now(),
        error,
    }
}

pub struct AuditLayer<A: AuditSink> {
    sink: Arc<A>,
    audit_get: bool,
    audit_subscribe: bool,
    audit_rpc: bool,
}

impl<A: AuditSink> AuditLayer<A> {
    /// New layer that audits PUT only (high-signal events).
    ///
    /// **Note**: `sink.record()` runs synchronously inside the PUT
    /// path. If your sink does blocking I/O (file write, syslog,
    /// HTTP), wrap it with [`MpscAuditSink::wrap`] first or use the
    /// [`AuditLayer::with_blocking_sink`] convenience constructor —
    /// otherwise gateway worker threads stall under sustained load.
    pub fn new(sink: A) -> Self {
        Self {
            sink: Arc::new(sink),
            audit_get: false,
            audit_subscribe: false,
            audit_rpc: false,
        }
    }
}

impl AuditLayer<MpscAuditSink> {
    /// Construct an audit layer where the user-supplied sink may
    /// block (file write, syslog, etc.). Wraps `inner` in an
    /// [`MpscAuditSink`] of `capacity`; the layer's `record()`
    /// becomes a non-blocking try_send and a background drainer task
    /// services the blocking I/O. Audit events past `capacity` are
    /// dropped (counted via [`MpscAuditSink::drops`]).
    pub fn with_blocking_sink<I: AuditSink>(capacity: usize, inner: I) -> Self {
        Self {
            sink: Arc::new(MpscAuditSink::wrap(capacity, inner)),
            audit_get: false,
            audit_subscribe: false,
            audit_rpc: false,
        }
    }
}

impl<A: AuditSink> AuditLayer<A> {
    /// Also emit an audit event on every GET. Off by default
    /// because GET frequency dominates real workloads (a Phoebus
    /// dashboard polls dozens per second).
    pub fn with_get(mut self) -> Self {
        self.audit_get = true;
        self
    }

    /// Also audit subscribe (monitor INIT). One event per
    /// subscriber connect — distinct from per-update events.
    pub fn with_subscribe(mut self) -> Self {
        self.audit_subscribe = true;
        self
    }

    /// Also audit RPC dispatch.
    pub fn with_rpc(mut self) -> Self {
        self.audit_rpc = true;
        self
    }
}

pub struct Audited<S, A> {
    inner: Arc<S>,
    sink: Arc<A>,
    audit_get: bool,
    audit_subscribe: bool,
    audit_rpc: bool,
}

impl<S: ChannelSource, A: AuditSink> Layer<S> for AuditLayer<A> {
    type Wrapped = Audited<S, A>;
    fn layer(self, inner: S) -> Audited<S, A> {
        Audited {
            inner: Arc::new(inner),
            sink: self.sink,
            audit_get: self.audit_get,
            audit_subscribe: self.audit_subscribe,
            audit_rpc: self.audit_rpc,
        }
    }
}

impl<S: ChannelSource, A: AuditSink> ChannelSource for Audited<S, A> {
    fn access(&self) -> &epics_pva_rs::server_native::source::AccessGate {
        self.inner.access()
    }
    async fn list_pvs(&self) -> Vec<String> {
        self.inner.list_pvs().await
    }
    async fn has_pv(&self, name: &str) -> bool {
        self.inner.has_pv(name).await
    }
    async fn get_introspection(&self, name: &str) -> Option<FieldDesc> {
        self.inner.get_introspection(name).await
    }
    async fn get_value(&self, name: &str) -> Option<PvField> {
        let result = self.inner.get_value(name).await;
        if self.audit_get {
            // GET has no error path here — None means "missing" not
            // "denied" — so we model it as Ok with empty error.
            let outcome: Result<(), String> = if result.is_some() {
                Ok(())
            } else {
                Err(format!("PV '{name}' not found"))
            };
            let mut ev = make_audit_event(name, "", "", &outcome);
            ev.event = AuditEventKind::Get;
            self.sink.record(ev);
        }
        result
    }
    async fn put_value(&self, name: &str, value: PvField) -> Result<(), String> {
        let result = self.inner.put_value(name, value).await;
        self.sink.record(make_audit_event(name, "", "", &result));
        result
    }
    // Round 43: typed PUT — emits a credential-aware audit row and
    // forwards through the inner's gate-enforced path.
    async fn put_value_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        value: PvField,
        ctx: ChannelContext,
    ) -> Result<(), String> {
        let pv = checked.pv_name().to_string();
        let user = ctx.account.clone();
        let host = ctx.host.clone();
        let result = self.inner.put_value_checked(checked, value, ctx).await;
        self.sink
            .record(make_audit_event(&pv, &user, &host, &result));
        result
    }
    // A BitSet-delta PUT records the same audit row shape as
    // `put_value_checked` (event kind Put, peer credentials) and
    // forwards through the inner's `put_delta_checked`. Without this
    // override the trait default would run get_value + merge +
    // put_value_checked on THIS layer: it would still audit (via the
    // put_value_checked above) but bypass the inner source's atomic
    // `put_delta_checked`, re-opening the concurrent-partial-PUT
    // lost-update window.
    async fn put_delta_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        desc: FieldDesc,
        changed: epics_pva_rs::proto::BitSet,
        delta: PvField,
        ctx: ChannelContext,
    ) -> Result<(), String> {
        let pv = checked.pv_name().to_string();
        let user = ctx.account.clone();
        let host = ctx.host.clone();
        let result = self
            .inner
            .put_delta_checked(checked, desc, changed, delta, ctx)
            .await;
        self.sink
            .record(make_audit_event(&pv, &user, &host, &result));
        result
    }
    // PROCESS is a record-state mutation — audit it with the same
    // row shape as PUT (`AuditEventKind::Process`), then forward.
    // Without this override the trait-default `process` would never
    // reach the inner source and no audit row would be emitted.
    async fn process(&self, name: &str) -> Result<(), String> {
        let result = self.inner.process(name).await;
        let mut ev = make_audit_event(name, "", "", &result);
        ev.event = AuditEventKind::Process;
        self.sink.record(ev);
        result
    }
    // Typed PROCESS — emits a credential-aware audit row (mirroring
    // `put_value_checked`) and forwards through the inner's
    // gate-enforced `process_checked`.
    async fn process_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Result<(), String> {
        let pv = checked.pv_name().to_string();
        let user = ctx.account.clone();
        let host = ctx.host.clone();
        let result = self.inner.process_checked(checked, ctx).await;
        let mut ev = make_audit_event(&pv, &user, &host, &result);
        ev.event = AuditEventKind::Process;
        self.sink.record(ev);
        result
    }
    async fn get_value_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Option<PvField> {
        let pv = checked.pv_name().to_string();
        let user = ctx.account.clone();
        let host = ctx.host.clone();
        let result = self.inner.get_value_checked(checked, ctx).await;
        if self.audit_get {
            let outcome: Result<(), String> = if result.is_some() {
                Ok(())
            } else {
                Err(format!("PV '{pv}' not found / denied"))
            };
            let mut ev = make_audit_event(&pv, &user, &host, &outcome);
            ev.event = AuditEventKind::Get;
            self.sink.record(ev);
        }
        result
    }
    async fn subscribe_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
    ) -> Option<mpsc::Receiver<PvField>> {
        let pv = checked.pv_name().to_string();
        let user = ctx.account.clone();
        let host = ctx.host.clone();
        let result = self.inner.subscribe_checked(checked, ctx).await;
        if self.audit_subscribe {
            let outcome: Result<(), String> = if result.is_some() {
                Ok(())
            } else {
                Err(format!("PV '{pv}' not subscribable"))
            };
            let mut ev = make_audit_event(&pv, &user, &host, &outcome);
            ev.event = AuditEventKind::Subscribe;
            self.sink.record(ev);
        }
        result
    }
    async fn is_writable(&self, name: &str) -> bool {
        self.inner.is_writable(name).await
    }
    async fn subscribe(&self, name: &str) -> Option<mpsc::Receiver<PvField>> {
        let result = self.inner.subscribe(name).await;
        if self.audit_subscribe {
            let outcome: Result<(), String> = if result.is_some() {
                Ok(())
            } else {
                Err(format!("PV '{name}' not subscribable"))
            };
            let mut ev = make_audit_event(name, "", "", &outcome);
            ev.event = AuditEventKind::Subscribe;
            self.sink.record(ev);
        }
        result
    }
    async fn subscribe_raw(&self, name: &str) -> Option<mpsc::Receiver<RawMonitorEvent>> {
        // Audit on subscribe_raw too, since the F-G12 zero-copy
        // path bypasses the typed `subscribe` and would otherwise
        // miss the audit event entirely.
        let result = self.inner.subscribe_raw(name).await;
        if self.audit_subscribe {
            let outcome: Result<(), String> = if result.is_some() {
                Ok(())
            } else {
                Err(format!("PV '{name}' not subscribable (raw)"))
            };
            let mut ev = make_audit_event(name, "", "", &outcome);
            ev.event = AuditEventKind::Subscribe;
            self.sink.record(ev);
        }
        result
    }
    // Round 43: typed raw MONITOR — populate audit row with peer
    // credentials and forward through the inner's gate-enforced
    // path.
    async fn subscribe_raw_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: epics_pva_rs::server_native::source::ChannelContext,
    ) -> Option<mpsc::Receiver<RawMonitorEvent>> {
        let pv = checked.pv_name().to_string();
        let user = ctx.account.clone();
        let host = ctx.host.clone();
        let result = self.inner.subscribe_raw_checked(checked, ctx).await;
        if self.audit_subscribe {
            let outcome: Result<(), String> = if result.is_some() {
                Ok(())
            } else {
                Err(format!("PV '{pv}' not subscribable (raw)"))
            };
            let mut ev = make_audit_event(&pv, &user, &host, &outcome);
            ev.event = AuditEventKind::Subscribe;
            self.sink.record(ev);
        }
        result
    }
    // BR-R14: forward the event-affecting-options MONITOR variants,
    // recording the same Subscribe audit row as `subscribe_*_checked`.
    // The audit layer is outermost, so it records the attempt even
    // when the inner gateway source rejects an unsupported option.
    async fn subscribe_checked_opts(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
        opts: epics_pva_rs::server_native::MonitorOptions,
    ) -> Option<mpsc::Receiver<PvField>> {
        let pv = checked.pv_name().to_string();
        let user = ctx.account.clone();
        let host = ctx.host.clone();
        let result = self.inner.subscribe_checked_opts(checked, ctx, opts).await;
        if self.audit_subscribe {
            let outcome: Result<(), String> = if result.is_some() {
                Ok(())
            } else {
                Err(format!("PV '{pv}' not subscribable"))
            };
            let mut ev = make_audit_event(&pv, &user, &host, &outcome);
            ev.event = AuditEventKind::Subscribe;
            self.sink.record(ev);
        }
        result
    }
    async fn subscribe_raw_checked_opts(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        ctx: ChannelContext,
        opts: epics_pva_rs::server_native::MonitorOptions,
    ) -> Option<mpsc::Receiver<RawMonitorEvent>> {
        let pv = checked.pv_name().to_string();
        let user = ctx.account.clone();
        let host = ctx.host.clone();
        let result = self
            .inner
            .subscribe_raw_checked_opts(checked, ctx, opts)
            .await;
        if self.audit_subscribe {
            let outcome: Result<(), String> = if result.is_some() {
                Ok(())
            } else {
                Err(format!("PV '{pv}' not subscribable (raw)"))
            };
            let mut ev = make_audit_event(&pv, &user, &host, &outcome);
            ev.event = AuditEventKind::Subscribe;
            self.sink.record(ev);
        }
        result
    }
    async fn rpc(
        &self,
        name: &str,
        request_desc: FieldDesc,
        request_value: PvField,
    ) -> Result<(FieldDesc, PvField), String> {
        let result = self.inner.rpc(name, request_desc, request_value).await;
        if self.audit_rpc {
            let outcome: Result<(), String> = match &result {
                Ok(_) => Ok(()),
                Err(e) => Err(e.clone()),
            };
            let mut ev = make_audit_event(name, "", "", &outcome);
            ev.event = AuditEventKind::Rpc;
            self.sink.record(ev);
        }
        result
    }
    // Round 43: typed RPC — populate audit row with peer
    // credentials and forward through the inner's gate-enforced
    // path.
    async fn rpc_checked(
        &self,
        checked: epics_pva_rs::server_native::source::AccessChecked,
        request_desc: FieldDesc,
        request_value: PvField,
        ctx: epics_pva_rs::server_native::source::ChannelContext,
    ) -> Result<(FieldDesc, PvField), String> {
        let pv = checked.pv_name().to_string();
        let user = ctx.account.clone();
        let host = ctx.host.clone();
        let result = self
            .inner
            .rpc_checked(checked, request_desc, request_value, ctx)
            .await;
        if self.audit_rpc {
            let outcome: Result<(), String> = match &result {
                Ok(_) => Ok(()),
                Err(e) => Err(e.clone()),
            };
            let mut ev = make_audit_event(&pv, &user, &host, &outcome);
            ev.event = AuditEventKind::Rpc;
            self.sink.record(ev);
        }
        result
    }
    fn notify_watermark_high(&self, name: &str) {
        self.inner.notify_watermark_high(name);
    }
    fn notify_watermark_low(&self, name: &str) {
        self.inner.notify_watermark_low(name);
    }
}

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

    #[test]
    fn pattern_matching() {
        assert!(matches_pattern("MOTOR:*", "MOTOR:VAL"));
        assert!(matches_pattern("*VAL", "MOTOR:VAL"));
        assert!(matches_pattern("EXACT", "EXACT"));
        assert!(!matches_pattern("EXACT", "EXACT2"));
        assert!(!matches_pattern("MOTOR:*", "OTHER:VAL"));
    }

    #[test]
    fn acl_allow_only() {
        let cfg = AclConfig {
            allow_only: vec!["BL10C:*".into()],
            ..Default::default()
        };
        assert!(cfg.allowed("BL10C:VG-01:PRESSURE"));
        assert!(!cfg.allowed("RFP:HV"));
    }

    #[test]
    fn acl_deny_overrides_allow() {
        let cfg = AclConfig {
            allow_only: vec!["MOTOR:*".into()],
            deny: vec!["MOTOR:JOG:*".into()],
            ..Default::default()
        };
        assert!(cfg.allowed("MOTOR:VAL"));
        assert!(!cfg.allowed("MOTOR:JOG:UP"));
        assert!(!cfg.allowed("OTHER:PV"));
    }

    // ── B7: regex ACL ────────────────────────────────────────────

    /// A regex deny entry is anchored at both ends, so `BL10C:.*`
    /// matches the same names a `BL10C:*` glob would.
    #[test]
    fn acl_regex_deny_anchored() {
        let cfg = AclConfig::default().deny_regex(r"BL10C:.*:HV").unwrap();
        assert!(!cfg.allowed("BL10C:RFP:HV"));
        assert!(!cfg.allowed("BL10C::HV"));
        // Anchored: the regex must match the WHOLE name.
        assert!(cfg.allowed("X:BL10C:RFP:HV"));
        assert!(cfg.allowed("BL10C:RFP:HV:SETPOINT"));
        assert!(cfg.allowed("BL10D:RFP:HV"));
    }

    /// A regex allow-only entry flips the policy to default-deny,
    /// exactly like a glob allow-only entry.
    #[test]
    fn acl_regex_allow_only_default_denies() {
        let cfg = AclConfig::default().allow_regex(r"(SR|BL)\d+:.*").unwrap();
        assert!(cfg.allowed("SR01:CURRENT"));
        assert!(cfg.allowed("BL10:SHUTTER"));
        assert!(!cfg.allowed("RFP:HV"));
        // Alternation stays anchored as ^(?:(SR|BL)\d+:.*)$ — a name
        // that merely contains a branch must not slip through.
        assert!(!cfg.allowed("X-SR01:CURRENT"));
    }

    /// Glob and regex patterns may be mixed in the same config;
    /// deny (glob or regex) always wins over allow (glob or regex).
    #[test]
    fn acl_glob_and_regex_mixed() {
        let cfg = AclConfig {
            allow_only: vec!["MOTOR:*".into()],
            ..Default::default()
        }
        .allow_regex(r"TEMP:\d+")
        .unwrap()
        .deny_regex(r"MOTOR:.*:JOG")
        .unwrap();
        // Glob allow.
        assert!(cfg.allowed("MOTOR:X:VAL"));
        // Regex allow.
        assert!(cfg.allowed("TEMP:42"));
        assert!(!cfg.allowed("TEMP:hot")); // \d+ requires digits
        // Regex deny beats glob allow.
        assert!(!cfg.allowed("MOTOR:X:JOG"));
        // Not in any allow list → default-deny.
        assert!(!cfg.allowed("RFP:HV"));
    }

    /// An invalid regex is reported at config-build time, not on the
    /// first PV check.
    #[test]
    fn acl_invalid_regex_rejected_at_build() {
        assert!(AclConfig::default().deny_regex(r"BL10C:[").is_err());
        assert!(AclConfig::default().allow_regex(r"(unclosed").is_err());
    }

    /// B7: a pathological operator-supplied ACL regex whose compiled
    /// program exceeds `ACL_REGEX_SIZE_LIMIT` must fail fast at build
    /// time with a bounded `regex::Error` rather than compiling with
    /// the crate's larger defaults. A bounded counted repetition of a
    /// large bounded inner repetition blows the program size well
    /// past 256 KiB while staying a syntactically valid pattern.
    #[test]
    fn acl_oversized_regex_rejected_by_size_limit() {
        // `(?:a{1000}){1000}` — syntactically valid, but the compiled
        // program is far larger than the 256 KiB ACL limit.
        let pathological = r"(?:a{1000}){1000}";

        // `compile_anchored` is the single funnel both `deny_regex`
        // and `allow_regex` route through — test it directly so the
        // failing error variant is observable (`AclConfig` is not
        // `Debug`, so `expect_err` on the builder is unavailable).
        match compile_anchored(pathological) {
            Err(regex::Error::CompiledTooBig(_)) => {}
            other => panic!("expected CompiledTooBig, got {other:?}"),
        }
        // Both ACL builder entry points reject it.
        assert!(
            AclConfig::default().deny_regex(pathological).is_err(),
            "oversized deny regex must be rejected"
        );
        assert!(
            AclConfig::default().allow_regex(pathological).is_err(),
            "oversized allow regex must be rejected"
        );

        // A realistic ACL pattern still compiles fine under the limit.
        assert!(compile_anchored(r"BL\d+C:.*:HV").is_ok());
        assert!(AclConfig::default().deny_regex(r"BL\d+C:.*:HV").is_ok());
    }

    /// The regex ACL still gates through the `AclLayer` wrapper on a
    /// real `ChannelSource`, not just the bare `AclConfig::allowed`.
    #[tokio::test]
    async fn acl_layer_applies_regex_via_channel_source() {
        use super::super::channel_cache::{ChannelCache, DEFAULT_CLEANUP_INTERVAL};
        use super::super::source::GatewayChannelSource;
        use epics_pva_rs::client::PvaClient;

        let client = Arc::new(PvaClient::builder().build());
        let cache = ChannelCache::new(client, DEFAULT_CLEANUP_INTERVAL);
        let inner = GatewayChannelSource::new(cache);

        let cfg = AclConfig::default().deny_regex(r"SECRET:.*").unwrap();
        let acl = AclLayer::new(cfg).layer(inner);

        // Denied name short-circuits at the layer — has_pv is false
        // and no upstream search is triggered.
        assert!(!acl.has_pv("SECRET:KEY").await);
        // put_value on a denied name returns the ACL error without
        // reaching the upstream.
        let err = acl
            .put_value(
                "SECRET:KEY",
                PvField::Scalar(epics_pva_rs::pvdata::ScalarValue::Double(0.0)),
            )
            .await
            .expect_err("regex-denied PUT must fail at the ACL layer");
        assert!(err.contains("denied"));
    }

    /// Audit-event classifier tags ACL/read-only error messages
    /// as Denied; other errors as Failed; Ok results as Ok.
    #[test]
    fn audit_event_classifies_results() {
        let denied = make_audit_event(
            "MOTOR:VAL",
            "alice",
            "host1",
            &Err("ACL: PV 'MOTOR:VAL' denied".into()),
        );
        assert_eq!(denied.result, AuditResult::Denied);
        assert!(!denied.error.is_empty());

        let read_only = make_audit_event(
            "MOTOR:VAL",
            "",
            "",
            &Err("read-only mode: PUT rejected".into()),
        );
        assert_eq!(read_only.result, AuditResult::Denied);

        let failed = make_audit_event(
            "MOTOR:VAL",
            "alice",
            "host1",
            &Err("upstream timeout".into()),
        );
        assert_eq!(failed.result, AuditResult::Failed);

        let ok = make_audit_event("MOTOR:VAL", "alice", "host1", &Ok(()));
        assert_eq!(ok.result, AuditResult::Ok);
        assert!(ok.error.is_empty());
    }

    // ── put_delta_checked forwarding through the wrappers ────────────

    use epics_pva_rs::pvdata::{ScalarType, ScalarValue};
    use epics_pva_rs::server_native::source::{AccessChecked, AccessGate};
    use std::sync::atomic::{AtomicBool, Ordering};

    /// Minimal `ChannelSource` stub that records which PUT method
    /// the wrapper routed a delta PUT to. `put_delta_checked` is
    /// overridden (atomic merge stand-in); the default
    /// `put_value_checked` chains through it. If a wrapper bypasses
    /// `put_delta_checked` and runs the non-atomic default merge, it
    /// lands in `put_value_checked` and `delta_reached` stays false.
    struct RecordingSource {
        delta_reached: Arc<AtomicBool>,
        value_reached: Arc<AtomicBool>,
        process_reached: Arc<AtomicBool>,
    }

    impl ChannelSource for RecordingSource {
        async fn list_pvs(&self) -> Vec<String> {
            vec!["X".into()]
        }
        async fn has_pv(&self, _name: &str) -> bool {
            true
        }
        async fn get_introspection(&self, _name: &str) -> Option<FieldDesc> {
            Some(FieldDesc::Scalar(ScalarType::Double))
        }
        async fn get_value(&self, _name: &str) -> Option<PvField> {
            Some(PvField::Scalar(ScalarValue::Double(0.0)))
        }
        async fn put_value(&self, _name: &str, _value: PvField) -> Result<(), String> {
            Ok(())
        }
        async fn put_value_checked(
            &self,
            _checked: AccessChecked,
            _value: PvField,
            _ctx: ChannelContext,
        ) -> Result<(), String> {
            self.value_reached.store(true, Ordering::SeqCst);
            Ok(())
        }
        async fn put_delta_checked(
            &self,
            _checked: AccessChecked,
            _desc: FieldDesc,
            _changed: epics_pva_rs::proto::BitSet,
            _delta: PvField,
            _ctx: ChannelContext,
        ) -> Result<(), String> {
            self.delta_reached.store(true, Ordering::SeqCst);
            Ok(())
        }
        // Inner PROCESS sink. The trait-default `process_checked`
        // applies the `allows_write()` gate then delegates here, so a
        // wrapper that correctly forwards `process_checked` lands in
        // this method and flips `process_reached`.
        async fn process(&self, _name: &str) -> Result<(), String> {
            self.process_reached.store(true, Ordering::SeqCst);
            Ok(())
        }
        async fn is_writable(&self, _name: &str) -> bool {
            true
        }
        async fn subscribe(&self, _name: &str) -> Option<mpsc::Receiver<PvField>> {
            None
        }
    }

    fn test_ctx() -> ChannelContext {
        ChannelContext {
            peer: "127.0.0.1:5075".parse().unwrap(),
            account: "alice".into(),
            method: "anonymous".into(),
            host: "host1".into(),
            authority: String::new(),
            roles: Vec::new(),
            pv_request: None,
        }
    }

    async fn checked_for(name: &str) -> AccessChecked {
        AccessGate::open()
            .check(name, "host1", "alice", "anonymous", "")
            .await
    }

    /// A delta PUT through the `Acl` wrapper must reach the inner
    /// source's `put_delta_checked` — not the non-atomic
    /// get+put_value_checked default merge.
    #[tokio::test]
    async fn acl_forwards_put_delta_checked_to_inner() {
        let delta_reached = Arc::new(AtomicBool::new(false));
        let value_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: delta_reached.clone(),
            value_reached: value_reached.clone(),
            process_reached: Arc::new(AtomicBool::new(false)),
        };
        let acl = AclLayer::new(AclConfig::default()).layer(inner);

        let mut changed = epics_pva_rs::proto::BitSet::new();
        changed.set(0);
        let res = acl
            .put_delta_checked(
                checked_for("X").await,
                FieldDesc::Scalar(ScalarType::Double),
                changed,
                PvField::Scalar(ScalarValue::Double(1.0)),
                test_ctx(),
            )
            .await;
        assert!(res.is_ok());
        assert!(
            delta_reached.load(Ordering::SeqCst),
            "Acl must route delta PUT to inner put_delta_checked"
        );
        assert!(
            !value_reached.load(Ordering::SeqCst),
            "Acl must NOT fall back to the non-atomic put_value_checked merge"
        );
    }

    /// An `Acl`-denied delta PUT must be rejected at the layer and
    /// never reach the inner source at all.
    #[tokio::test]
    async fn acl_denied_put_delta_checked_short_circuits() {
        let delta_reached = Arc::new(AtomicBool::new(false));
        let value_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: delta_reached.clone(),
            value_reached: value_reached.clone(),
            process_reached: Arc::new(AtomicBool::new(false)),
        };
        let cfg = AclConfig::default().deny_regex(r"SECRET:.*").unwrap();
        let acl = AclLayer::new(cfg).layer(inner);

        let mut changed = epics_pva_rs::proto::BitSet::new();
        changed.set(0);
        let err = acl
            .put_delta_checked(
                checked_for("SECRET:KEY").await,
                FieldDesc::Scalar(ScalarType::Double),
                changed,
                PvField::Scalar(ScalarValue::Double(1.0)),
                test_ctx(),
            )
            .await
            .expect_err("ACL-denied delta PUT must fail at the layer");
        assert!(err.contains("denied"));
        assert!(!delta_reached.load(Ordering::SeqCst));
        assert!(!value_reached.load(Ordering::SeqCst));
    }

    /// A delta PUT through the `Audited` wrapper must reach the
    /// inner's `put_delta_checked` and emit one `Put` audit row
    /// carrying the peer credentials.
    #[tokio::test]
    async fn audited_forwards_put_delta_checked_and_records() {
        let delta_reached = Arc::new(AtomicBool::new(false));
        let value_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: delta_reached.clone(),
            value_reached: value_reached.clone(),
            process_reached: Arc::new(AtomicBool::new(false)),
        };
        let events: Arc<std::sync::Mutex<Vec<AuditEvent>>> =
            Arc::new(std::sync::Mutex::new(Vec::new()));
        let events_sink = events.clone();
        let audited = AuditLayer::new(ClosureAudit(move |ev| {
            events_sink.lock().unwrap().push(ev);
        }))
        .layer(inner);

        let mut changed = epics_pva_rs::proto::BitSet::new();
        changed.set(0);
        let res = audited
            .put_delta_checked(
                checked_for("X").await,
                FieldDesc::Scalar(ScalarType::Double),
                changed,
                PvField::Scalar(ScalarValue::Double(1.0)),
                test_ctx(),
            )
            .await;
        assert!(res.is_ok());
        assert!(
            delta_reached.load(Ordering::SeqCst),
            "Audited must route delta PUT to inner put_delta_checked"
        );
        assert!(
            !value_reached.load(Ordering::SeqCst),
            "Audited must NOT fall back to the non-atomic put_value_checked merge"
        );
        let recorded = events.lock().unwrap();
        assert_eq!(recorded.len(), 1, "exactly one audit row per delta PUT");
        assert_eq!(recorded[0].event, AuditEventKind::Put);
        assert_eq!(recorded[0].result, AuditResult::Ok);
        assert_eq!(recorded[0].pv, "X");
        assert_eq!(recorded[0].user, "alice");
        assert_eq!(recorded[0].host, "host1");
    }

    /// A delta PUT through the `ReadOnly` wrapper must be rejected
    /// at the layer and never reach the inner source.
    #[tokio::test]
    async fn read_only_rejects_put_delta_checked() {
        let delta_reached = Arc::new(AtomicBool::new(false));
        let value_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: delta_reached.clone(),
            value_reached: value_reached.clone(),
            process_reached: Arc::new(AtomicBool::new(false)),
        };
        let ro = ReadOnlyLayer.layer(inner);

        let mut changed = epics_pva_rs::proto::BitSet::new();
        changed.set(0);
        let err = ro
            .put_delta_checked(
                checked_for("X").await,
                FieldDesc::Scalar(ScalarType::Double),
                changed,
                PvField::Scalar(ScalarValue::Double(1.0)),
                test_ctx(),
            )
            .await
            .expect_err("read-only delta PUT must be rejected");
        assert!(err.contains("read-only"));
        assert!(!delta_reached.load(Ordering::SeqCst));
        assert!(!value_reached.load(Ordering::SeqCst));
    }

    // ── process / process_checked forwarding through the wrappers ────

    /// A typed PROCESS through the `Acl` wrapper (allowed name) must
    /// reach the inner source's PROCESS path. Pre-fix the trait
    /// default `process_checked` → `process` (`Ok(())`) ran on the
    /// `Acl` layer itself and never forwarded.
    #[tokio::test]
    async fn acl_forwards_process_checked_to_inner() {
        let process_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: Arc::new(AtomicBool::new(false)),
            value_reached: Arc::new(AtomicBool::new(false)),
            process_reached: process_reached.clone(),
        };
        let acl = AclLayer::new(AclConfig::default()).layer(inner);

        let res = acl
            .process_checked(checked_for("X").await, test_ctx())
            .await;
        assert!(res.is_ok());
        assert!(
            process_reached.load(Ordering::SeqCst),
            "Acl must route PROCESS to the inner source's process_checked"
        );
    }

    /// An `Acl`-denied PROCESS must be rejected at the layer and never
    /// reach the inner source.
    #[tokio::test]
    async fn acl_denied_process_checked_short_circuits() {
        let process_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: Arc::new(AtomicBool::new(false)),
            value_reached: Arc::new(AtomicBool::new(false)),
            process_reached: process_reached.clone(),
        };
        let cfg = AclConfig::default().deny_regex(r"SECRET:.*").unwrap();
        let acl = AclLayer::new(cfg).layer(inner);

        let err = acl
            .process_checked(checked_for("SECRET:KEY").await, test_ctx())
            .await
            .expect_err("ACL-denied PROCESS must fail at the layer");
        assert!(err.contains("denied"));
        assert!(
            !process_reached.load(Ordering::SeqCst),
            "ACL-denied PROCESS must not reach the inner source"
        );
    }

    /// A PROCESS through the `ReadOnly` wrapper must be rejected at
    /// the layer (WRITE-class op) and never reach the inner source.
    #[tokio::test]
    async fn read_only_rejects_process_checked() {
        let process_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: Arc::new(AtomicBool::new(false)),
            value_reached: Arc::new(AtomicBool::new(false)),
            process_reached: process_reached.clone(),
        };
        let ro = ReadOnlyLayer.layer(inner);

        let err = ro
            .process_checked(checked_for("X").await, test_ctx())
            .await
            .expect_err("read-only PROCESS must be rejected");
        assert!(err.contains("read-only"));
        assert!(
            !process_reached.load(Ordering::SeqCst),
            "read-only PROCESS must not reach the inner source"
        );
        // The ctx-less `process` is rejected the same way.
        let err = ro
            .process("X")
            .await
            .expect_err("read-only ctx-less PROCESS must be rejected");
        assert!(err.contains("read-only"));
        assert!(!process_reached.load(Ordering::SeqCst));
    }

    /// A PROCESS through the `Audited` wrapper must reach the inner
    /// source and emit exactly one `Process` audit row carrying the
    /// peer credentials.
    #[tokio::test]
    async fn audited_forwards_process_checked_and_records() {
        let process_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: Arc::new(AtomicBool::new(false)),
            value_reached: Arc::new(AtomicBool::new(false)),
            process_reached: process_reached.clone(),
        };
        let events: Arc<std::sync::Mutex<Vec<AuditEvent>>> =
            Arc::new(std::sync::Mutex::new(Vec::new()));
        let events_sink = events.clone();
        let audited = AuditLayer::new(ClosureAudit(move |ev| {
            events_sink.lock().unwrap().push(ev);
        }))
        .layer(inner);

        let res = audited
            .process_checked(checked_for("X").await, test_ctx())
            .await;
        assert!(res.is_ok());
        assert!(
            process_reached.load(Ordering::SeqCst),
            "Audited must route PROCESS to the inner source's process_checked"
        );
        let recorded = events.lock().unwrap();
        assert_eq!(recorded.len(), 1, "exactly one audit row per PROCESS");
        assert_eq!(recorded[0].event, AuditEventKind::Process);
        assert_eq!(recorded[0].result, AuditResult::Ok);
        assert_eq!(recorded[0].pv, "X");
        assert_eq!(recorded[0].user, "alice");
        assert_eq!(recorded[0].host, "host1");
    }

    /// The full production layer stack `Audited(Acl(inner))` (what
    /// `PvaGateway::start` builds) must forward a PROCESS all the way
    /// to the inner source — proving the fix is effective in the real
    /// layered deployment, not just for a single wrapper. Pre-fix the
    /// PROCESS hit `Audited`'s trait-default `process_checked` →
    /// `process` (`Ok(())`) and never reached `Acl`, let alone the
    /// inner source.
    #[tokio::test]
    async fn layered_audited_acl_forwards_process_to_inner() {
        let process_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: Arc::new(AtomicBool::new(false)),
            value_reached: Arc::new(AtomicBool::new(false)),
            process_reached: process_reached.clone(),
        };
        let events: Arc<std::sync::Mutex<Vec<AuditEvent>>> =
            Arc::new(std::sync::Mutex::new(Vec::new()));
        let events_sink = events.clone();
        let acl = AclLayer::new(AclConfig::default()).layer(inner);
        let layered = AuditLayer::new(ClosureAudit(move |ev| {
            events_sink.lock().unwrap().push(ev);
        }))
        .layer(acl);

        let res = layered
            .process_checked(checked_for("X").await, test_ctx())
            .await;
        assert!(res.is_ok());
        assert!(
            process_reached.load(Ordering::SeqCst),
            "PROCESS through Audited(Acl(inner)) must reach the inner source"
        );
        assert_eq!(
            events.lock().unwrap()[0].event,
            AuditEventKind::Process,
            "the layered PROCESS must still be audited"
        );
    }

    /// In the layered stack, an ACL deny still short-circuits a
    /// PROCESS at the `Acl` layer (the audit row records the denial).
    #[tokio::test]
    async fn layered_audited_acl_denies_process() {
        let process_reached = Arc::new(AtomicBool::new(false));
        let inner = RecordingSource {
            delta_reached: Arc::new(AtomicBool::new(false)),
            value_reached: Arc::new(AtomicBool::new(false)),
            process_reached: process_reached.clone(),
        };
        let events: Arc<std::sync::Mutex<Vec<AuditEvent>>> =
            Arc::new(std::sync::Mutex::new(Vec::new()));
        let events_sink = events.clone();
        let cfg = AclConfig::default().deny_regex(r"SECRET:.*").unwrap();
        let acl = AclLayer::new(cfg).layer(inner);
        let layered = AuditLayer::new(ClosureAudit(move |ev| {
            events_sink.lock().unwrap().push(ev);
        }))
        .layer(acl);

        let err = layered
            .process_checked(checked_for("SECRET:KEY").await, test_ctx())
            .await
            .expect_err("ACL-denied PROCESS must fail through the layered stack");
        assert!(err.contains("denied"));
        assert!(
            !process_reached.load(Ordering::SeqCst),
            "ACL-denied PROCESS must not reach the inner source"
        );
        let recorded = events.lock().unwrap();
        assert_eq!(recorded[0].event, AuditEventKind::Process);
        assert_eq!(recorded[0].result, AuditResult::Denied);
    }

    /// BR-R14 regression: minimal source that records whether its
    /// event-affecting-options MONITOR variant was reached.
    struct OptsRecordingSource {
        opts_reached: Arc<AtomicBool>,
    }

    impl ChannelSource for OptsRecordingSource {
        async fn list_pvs(&self) -> Vec<String> {
            vec!["X".into()]
        }
        async fn has_pv(&self, _name: &str) -> bool {
            true
        }
        async fn get_introspection(&self, _name: &str) -> Option<FieldDesc> {
            Some(FieldDesc::Scalar(ScalarType::Double))
        }
        async fn get_value(&self, _name: &str) -> Option<PvField> {
            Some(PvField::Scalar(ScalarValue::Double(0.0)))
        }
        async fn put_value(&self, _name: &str, _value: PvField) -> Result<(), String> {
            Ok(())
        }
        async fn is_writable(&self, _name: &str) -> bool {
            false
        }
        async fn subscribe(&self, _name: &str) -> Option<mpsc::Receiver<PvField>> {
            None
        }
        async fn subscribe_checked_opts(
            &self,
            _checked: AccessChecked,
            _ctx: ChannelContext,
            _opts: epics_pva_rs::server_native::MonitorOptions,
        ) -> Option<mpsc::Receiver<PvField>> {
            self.opts_reached.store(true, Ordering::SeqCst);
            None
        }
        async fn subscribe_raw_checked_opts(
            &self,
            _checked: AccessChecked,
            _ctx: ChannelContext,
            _opts: epics_pva_rs::server_native::MonitorOptions,
        ) -> Option<mpsc::Receiver<RawMonitorEvent>> {
            self.opts_reached.store(true, Ordering::SeqCst);
            None
        }
    }

    /// BR-R14: the `Audit` / `ReadOnly` / `Acl` middleware wrappers
    /// must forward `subscribe_checked_opts` /
    /// `subscribe_raw_checked_opts` to the inner source. Pre-fix the
    /// wrappers inherited the trait default, which drops `opts` and
    /// routes through `subscribe_checked` — so the gateway source (the
    /// innermost layer) never saw the downstream's event-affecting
    /// monitor options and could not reject an unsupported set.
    #[tokio::test]
    async fn middleware_layers_forward_subscribe_opts_to_inner() {
        use epics_pva_rs::server_native::MonitorOptions;

        for raw in [false, true] {
            let opts_reached = Arc::new(AtomicBool::new(false));
            let inner = OptsRecordingSource {
                opts_reached: opts_reached.clone(),
            };
            // Production gateway stack shape: Audit( ReadOnly( Acl( inner ) ) ).
            let acl = AclLayer::new(AclConfig::default()).layer(inner);
            let read_only = ReadOnlyLayer.layer(acl);
            let layered = AuditLayer::new(NoopAudit).layer(read_only);

            let opts = MonitorOptions {
                pipeline: false,
                queue_size: None,
                server_filter: true,
            };
            if raw {
                let _ = layered
                    .subscribe_raw_checked_opts(checked_for("X").await, test_ctx(), opts)
                    .await;
            } else {
                let _ = layered
                    .subscribe_checked_opts(checked_for("X").await, test_ctx(), opts)
                    .await;
            }
            assert!(
                opts_reached.load(Ordering::SeqCst),
                "middleware stack must forward subscribe{}_checked_opts to the inner source",
                if raw { "_raw" } else { "" },
            );
        }
    }
}