act-runtime 0.13.1

//! The audit layer.
//!
//! Rollup requires state accumulated across events within a span, which a
//! `FormatEvent` implementation cannot hold — so this is a full `Layer` that
//! keeps per-span state in span extensions:
//!
//! * `on_new_span`  — capture the envelope fields, install an empty `Rollup`
//! * `on_record`    — pick up `outcome` / `duration_ms` recorded at finish
//! * `on_event`     — print exceptions now, fold allows into the parent span
//! * `on_close`     — render and flush the rollup line

use std::io::Write;
use std::sync::Mutex;

use tracing::field::{Field, Visit};
use tracing::{Event, Subscriber, span};
use tracing_subscriber::layer::{Context, Layer};
use tracing_subscriber::registry::LookupSpan;

use crate::audit::TARGET_AUDIT;
use crate::audit::record::{
    Actor, CapDecisionRecord, CeilingClassRecord, CredentialIssueRecord, Decision4, attr,
};
use crate::audit::render::{
    Rollup, SpanFields, render_credential_issue, render_declared_ask_blocked_warning,
    render_declared_ungranted_warning, render_exception, render_header, render_rollup,
};

/// Name of the tool-call envelope span, set by `emit::tool_call_span`.
const SPAN_TOOL_CALL: &str = "act.tool_call";
/// Name of the instantiation envelope span, set by `emit::instantiation_span`.
const SPAN_INSTANTIATION: &str = "act.instantiation";

/// Default cap on distinct rollup groups per tool call. Chosen to comfortably
/// cover a well-behaved component; past it, new groups collapse into a count.
pub const DEFAULT_ROLLUP_CAP: usize = 64;

/// How much the operator wants to see.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Detail {
    /// Exceptions immediately; allows summarised per tool call.
    Rollup,
    /// Every operation, plus the summary.
    Full,
}

/// Where rendered lines go. Abstracted so tests can capture them.
pub trait AuditWriter: Send + Sync + 'static {
    fn write_line(&self, line: &str);
}

/// The production sink: stderr, line-buffered, failures ignored.
pub struct StderrWriter {
    inner: Mutex<std::io::Stderr>,
}

impl Default for StderrWriter {
    fn default() -> Self {
        Self {
            inner: Mutex::new(std::io::stderr()),
        }
    }
}

impl AuditWriter for StderrWriter {
    fn write_line(&self, line: &str) {
        // A closed or full stderr degrades to silence. Audit must never
        // affect a decision, so nothing here can fail upward.
        if let Ok(mut w) = self.inner.lock() {
            let _ = writeln!(w, "{line}");
            let _ = w.flush();
        }
    }
}

pub struct AuditLayer<W> {
    writer: W,
    detail: Detail,
    rollup_cap: usize,
}

impl AuditLayer<StderrWriter> {
    pub fn stderr(detail: Detail) -> Self {
        Self::new(StderrWriter::default(), detail)
    }
}

impl<W: AuditWriter> AuditLayer<W> {
    pub fn new(writer: W, detail: Detail) -> Self {
        Self {
            writer,
            detail,
            rollup_cap: DEFAULT_ROLLUP_CAP,
        }
    }

    /// Never let a rendering or writing fault escape into enforcement.
    ///
    /// Takes a **closure**, not a `String`, deliberately: an argument would be
    /// evaluated before `catch_unwind` is entered, so a panic while rendering
    /// would unwind through the layer — and rendering handles guest-chosen
    /// values. The audit path must not be able to take the host down.
    fn emit(&self, render: impl FnOnce() -> String) {
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            self.writer.write_line(&render());
        }));
    }
}

/// Collects the envelope fields declared by `emit::tool_call_span`.
#[derive(Default)]
struct SpanVisitor(SpanFields);

impl Visit for SpanVisitor {
    fn record_str(&mut self, f: &Field, v: &str) {
        match f.name() {
            n if n == attr::COMPONENT_REF => self.0.component_ref = v.to_string(),
            n if n == attr::COMPONENT_DIGEST => self.0.digest = v.to_string(),
            n if n == attr::TOOL_NAME => self.0.tool = v.to_string(),
            n if n == attr::TOOL_ARGS_SHA256 => self.0.args_sha256 = v.to_string(),
            // Empty means --audit-args was not set for this call — same
            // "absent means empty string on the wire" convention SESSION_ID
            // uses below, so render_rollup's `Option` check can tell "no
            // value recorded" apart from "recorded as an empty string".
            n if n == attr::TOOL_ARGS && !v.is_empty() => self.0.args_json = Some(v.to_string()),
            // AGENT_ID, TRACE_PARENT and TRACE_STATE are captured onto the
            // span but stay unrendered by design; only REQUEST_ID reaches a
            // line, so an operator can join it back to a client log line.
            n if n == attr::REQUEST_ID => self.0.request_id = v.to_string(),
            n if n == attr::TRANSPORT => self.0.transport = v.to_string(),
            n if n == attr::OUTCOME => self.0.outcome = v.to_string(),
            n if n == attr::SESSION_ID && !v.is_empty() => {
                self.0.session_id = Some(v.to_string());
            }
            _ => {}
        }
    }

    fn record_u64(&mut self, f: &Field, v: u64) {
        if f.name() == attr::DURATION_MS {
            self.0.duration_ms = v;
        }
    }

    fn record_debug(&mut self, f: &Field, v: &dyn std::fmt::Debug) {
        // Every field here is recorded with `%value` (Display), which
        // tracing routes through `record_debug` with a wrapper whose `Debug`
        // impl just forwards to `Display` — so `{v:?}` is already the plain,
        // unquoted value. Do not strip quotes here: a guest-chosen value may
        // legitimately start or end with one, and trimming it corrupts data.
        let s = format!("{v:?}");
        self.record_str(f, &s);
    }
}

/// Collects one capability-decision event — or one ceiling-class event, from
/// an instantiation span — back into a record. The two share a visitor
/// because both carry `act.capability.id`; `into_record` / the `declared`
/// field is how the layer tells them apart (see `on_event`).
#[derive(Default)]
struct EventVisitor {
    cap_id: String,
    key: String,
    action: String,
    decision: String,
    mode: String,
    actor: String,
    reason: String,
    rule: String,
    declared: bool,
    never_rollup: bool,
    has_prompt_channel: bool,
    /// Present on a credential-issue event and on nothing else — see
    /// `on_event`, which branches on it before anything else.
    ///
    /// An `Option`, not a `String`, and the difference is load-bearing: the
    /// kind comes straight off the stored record and nothing validates it, so
    /// a record written with `kind: ""` is served to the guest all the same.
    /// Branching on emptiness would drop that event through both other
    /// branches too, and a secret would cross the sandbox boundary with no
    /// audit line at all — the one thing this record exists to make
    /// impossible. Presence of the field is the signal; its value is not.
    credential_kind: Option<String>,
    /// Carried on the credential-issue event itself rather than read off an
    /// enclosing span, so the record identifies itself without depending on
    /// span context.
    component_ref: String,
    session_id: String,
}

impl Visit for EventVisitor {
    fn record_str(&mut self, f: &Field, v: &str) {
        match f.name() {
            n if n == attr::CAPABILITY_ID => self.cap_id = v.to_string(),
            n if n == attr::RESOURCE_KEY => self.key = v.to_string(),
            n if n == attr::RESOURCE_ACTION => self.action = v.to_string(),
            n if n == attr::DECISION => self.decision = v.to_string(),
            n if n == attr::POLICY_MODE => self.mode = v.to_string(),
            n if n == attr::CREDENTIAL_KIND => self.credential_kind = Some(v.to_string()),
            n if n == attr::COMPONENT_REF => self.component_ref = v.to_string(),
            n if n == attr::SESSION_ID => self.session_id = v.to_string(),
            n if n == attr::POLICY_ACTOR => self.actor = v.to_string(),
            n if n == attr::POLICY_REASON => self.reason = v.to_string(),
            n if n == attr::POLICY_RULE => self.rule = v.to_string(),
            _ => {}
        }
    }

    fn record_bool(&mut self, f: &Field, v: bool) {
        // `act.capability.declared`, `act.consent.prompt_channel` and
        // `act.decision.never_rollup` are the only bool fields this crate
        // emits. Without this override, tracing's default `Visit::record_bool`
        // routes a bool to `record_debug`, whose output ("true"/"false")
        // doesn't match any `record_str` arm above — the field would be
        // silently dropped and neither instantiation warning could ever fire.
        match f.name() {
            n if n == attr::CAPABILITY_DECLARED => self.declared = v,
            n if n == attr::CONSENT_PROMPT_CHANNEL => self.has_prompt_channel = v,
            n if n == attr::NEVER_ROLLUP => self.never_rollup = v,
            _ => {}
        }
    }

    fn record_debug(&mut self, f: &Field, v: &dyn std::fmt::Debug) {
        // See the identical comment on `SpanVisitor::record_debug`: these
        // fields all arrive as `%value` and are already unquoted.
        let s = format!("{v:?}");
        self.record_str(f, &s);
    }
}

impl EventVisitor {
    fn into_record(self) -> Option<CapDecisionRecord> {
        if self.cap_id.is_empty() || self.decision.is_empty() {
            return None;
        }
        let decision = match self.decision.as_str() {
            "allow" => Decision4::Allow,
            "deny" => Decision4::Deny,
            "ask-allow" => Decision4::AskAllow,
            "ask-deny" => Decision4::AskDeny,
            _ => return None,
        };
        let actor = match self.actor.as_str() {
            "user" => Actor::User,
            "policy" => Actor::Policy,
            _ => Actor::Static,
        };
        Some(CapDecisionRecord {
            cap_id: self.cap_id,
            key: self.key,
            action: self.action,
            decision,
            mode: self.mode,
            actor,
            reason: (!self.reason.is_empty()).then_some(self.reason),
            rule: (!self.rule.is_empty()).then_some(self.rule),
            never_rollup: self.never_rollup,
        })
    }
}

struct SpanState {
    fields: SpanFields,
    rollup: Rollup,
}

/// Per-instantiation-span state: the envelope identity plus every capability
/// class resolved for this component load, collected as `emit_ceiling_class`
/// events arrive.
struct InstantiationState {
    component_ref: String,
    digest: String,
    classes: Vec<CeilingClassRecord>,
}

impl<S, W> Layer<S> for AuditLayer<W>
where
    S: Subscriber + for<'a> LookupSpan<'a>,
    W: AuditWriter,
{
    fn on_new_span(&self, attrs: &span::Attributes<'_>, id: &span::Id, ctx: Context<'_, S>) {
        if attrs.metadata().target() != TARGET_AUDIT {
            return;
        }
        let Some(span) = ctx.span(id) else { return };
        let mut v = SpanVisitor::default();
        attrs.record(&mut v);
        // Two envelope kinds share this target: a tool call gets the rollup
        // state it always had; an instantiation gets a plain vec of ceiling
        // classes. Branching on the span *name* (not just target) matters —
        // every audit-target span used to get a tool-call `SpanState`
        // unconditionally, which would make an instantiation span render as
        // a bogus tool call once it started reaching this layer at all.
        match attrs.metadata().name() {
            SPAN_TOOL_CALL => {
                span.extensions_mut().insert(SpanState {
                    fields: v.0,
                    rollup: Rollup::new(self.rollup_cap),
                });
            }
            SPAN_INSTANTIATION => {
                span.extensions_mut().insert(InstantiationState {
                    component_ref: v.0.component_ref,
                    digest: v.0.digest,
                    classes: Vec::new(),
                });
            }
            _ => {}
        }
    }

    fn on_record(&self, id: &span::Id, values: &span::Record<'_>, ctx: Context<'_, S>) {
        let Some(span) = ctx.span(id) else { return };
        let mut ext = span.extensions_mut();
        let Some(state) = ext.get_mut::<SpanState>() else {
            return;
        };
        let mut v = SpanVisitor(std::mem::take(&mut state.fields));
        values.record(&mut v);
        state.fields = v.0;
    }

    fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) {
        if event.metadata().target() != TARGET_AUDIT {
            return;
        }
        let mut v = EventVisitor::default();
        event.record(&mut v);

        // A credential-issue record (from `emit_credential_issue`) carries
        // neither a capability id nor a decision, so both branches below
        // would drop it. It is checked first, on the presence of the one
        // field no other audit event emits — presence, not a non-empty
        // value, so an unvalidated `kind` cannot make a secret cross
        // unrecorded.
        //
        // Printed unconditionally, under `Detail::Rollup` too: a per-run
        // count of "credentials issued" would not answer the only question
        // an operator has here, which is *which* ones.
        if let Some(kind) = v.credential_kind {
            let record = CredentialIssueRecord {
                component_ref: v.component_ref,
                session_id: v.session_id,
                key: v.key,
                kind,
            };
            self.emit(|| render_credential_issue(&record));
            return;
        }

        // A ceiling-class record (from `emit_ceiling_class`, inside an
        // instantiation span) carries a capability id but no `act.decision`
        // — a capability decision always carries both. That's the only
        // signal available at this point to tell the two event shapes
        // apart, so check it before falling through to `into_record`, which
        // requires a decision and would otherwise just drop this event.
        if !v.cap_id.is_empty() && v.decision.is_empty() {
            let record = CeilingClassRecord {
                cap_id: v.cap_id,
                mode: v.mode,
                declared: v.declared,
                has_prompt_channel: v.has_prompt_channel,
            };
            // Walk the enclosing spans looking for the instantiation span's
            // state; stop at the first match. A `for` loop, not
            // `Iterator::any`, because there's no boolean result anyone
            // reads — this is a search-and-push, not a predicate.
            for span in ctx.event_scope(event).into_iter().flatten() {
                if let Some(state) = span.extensions_mut().get_mut::<InstantiationState>() {
                    state.classes.push(record);
                    break;
                }
            }
            return;
        }

        let Some(record) = v.into_record() else {
            return;
        };

        // I2: a consent (semantic-class) decision must never fold into the
        // rollup, even when it is an Allow — same reasoning
        // `render_credential_issue`'s doc gives for a credential issue: there
        // are few of these, each is a distinct consequential act, and *which
        // subject* is the whole content of the decision. Treating
        // `never_rollup` as an exception here, alongside a real Deny/Ask,
        // both prints it immediately and (via the `is_exception()` guard
        // below) keeps it out of the fold.
        if record.decision.is_exception() || record.never_rollup || self.detail == Detail::Full {
            self.emit(|| render_exception(&record));
        }
        if !record.decision.is_exception() && !record.never_rollup {
            // Fold into the nearest enclosing tool-call span, if there is
            // one. `SpanState` is installed only on TARGET_AUDIT spans, so a
            // plain (non-audit) span nested between the event and the tool
            // call would make a direct parent lookup miss it — walk the
            // whole scope instead of just the immediate parent. A decision
            // fired outside any call (instantiation) has nowhere to roll up,
            // so it is printed instead of dropped.
            let folded = ctx.event_scope(event).is_some_and(|mut scope| {
                scope.any(|span| match span.extensions_mut().get_mut::<SpanState>() {
                    Some(state) => {
                        state
                            .rollup
                            .add(&record.cap_id, &record.action, record.rule.as_deref());
                        true
                    }
                    None => false,
                })
            });
            // Carries the instantiation-time guarantee: an allow with
            // nowhere to fold (no enclosing tool-call span) would otherwise
            // be silently dropped under Detail::Rollup. Detail::Full already
            // printed it above, so this only fires under Detail::Rollup.
            if !folded && self.detail != Detail::Full {
                self.emit(|| render_exception(&record));
            }
        }
    }

    fn on_close(&self, id: span::Id, ctx: Context<'_, S>) {
        let Some(span) = ctx.span(&id) else { return };
        let mut ext = span.extensions_mut();
        if let Some(state) = ext.remove::<SpanState>() {
            drop(ext);
            self.emit(|| render_rollup(&state.fields, &state.rollup));
            return;
        }
        let Some(state) = ext.remove::<InstantiationState>() else {
            return;
        };
        drop(ext);
        let modes: Vec<(String, String)> = state
            .classes
            .iter()
            .map(|c| (c.cap_id.clone(), c.mode.clone()))
            .collect();
        self.emit(|| render_header(&state.component_ref, &state.digest, &modes));
        // Only a class the component actually declared, that still resolved
        // to deny, is worth a warning — every undeclared class also resolves
        // to deny, and flagging all of those would bury the one signal an
        // operator needs: a capability the component asked for that nothing
        // granted.
        let ungranted: Vec<String> = state
            .classes
            .iter()
            .filter(|c| c.declared && c.mode == "deny")
            .map(|c| c.cap_id.clone())
            .collect();
        if !ungranted.is_empty() {
            self.emit(|| render_declared_ungranted_warning(&ungranted));
        }
        // A declared class configured as `ask` is not actually reachable
        // when this run has no prompt channel at all (headless / ACT-HTTP):
        // every access degrades to deny before a human is ever asked. The
        // header keeps showing `ask` — that is genuinely the configured
        // policy — but this second, distinct warning names the outcome the
        // operator will actually see, the same way the deny warning above
        // does for a hard deny. A declared `ask` class backed by a real
        // prompt channel (TTY / MCP elicitation) is not warned about here:
        // nothing has been refused yet at instantiation time.
        let ask_blocked: Vec<String> = state
            .classes
            .iter()
            .filter(|c| c.declared && c.mode == "ask" && !c.has_prompt_channel)
            .map(|c| c.cap_id.clone())
            .collect();
        if !ask_blocked.is_empty() {
            self.emit(|| render_declared_ask_blocked_warning(&ask_blocked));
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};
    use std::time::Duration;

    use tracing_subscriber::prelude::*;

    use super::*;
    use crate::audit::emit::{
        emit_cap_decision, emit_ceiling_class, emit_credential_issue, finish_tool_call,
        instantiation_span, tool_call_span,
    };
    use crate::audit::record::*;

    #[derive(Clone, Default)]
    struct TestWriter(Arc<Mutex<Vec<String>>>);

    impl AuditWriter for TestWriter {
        fn write_line(&self, line: &str) {
            self.0.lock().unwrap().push(line.to_string());
        }
    }

    fn start() -> ToolCallStart {
        ToolCallStart {
            component_ref: "python-eval@0.16.0".into(),
            digest: "1f3a9c4e5d6b7a8c".into(),
            tool: "run_python".into(),
            args_sha256: "9e21c4aa".into(),
            args_json: None,
            session_id: None,
            transport: Transport::Cli,
            agent_id: None,
            request_id: "req-1".into(),
            traceparent: None,
            tracestate: None,
        }
    }

    fn allow(action: &str, rule: &str) -> CapDecisionRecord {
        CapDecisionRecord {
            cap_id: "wasi:filesystem".into(),
            key: "/data/app.db".into(),
            action: action.into(),
            decision: Decision4::Allow,
            mode: "allowlist".into(),
            actor: Actor::Static,
            reason: None,
            rule: Some(rule.into()),
            never_rollup: false,
        }
    }

    fn deny() -> CapDecisionRecord {
        CapDecisionRecord {
            cap_id: "wasi:http".into(),
            key: "evil.example.com:443".into(),
            action: "GET".into(),
            decision: Decision4::Deny,
            mode: "ask".into(),
            actor: Actor::Static,
            reason: Some("outside ceiling".into()),
            rule: None,
            never_rollup: false,
        }
    }

    /// A consent (semantic-class) decision — `never_rollup: true`. Used by
    /// the I2 rollup-exemption tests below to prove the layer treats it like
    /// a credential issue: printed immediately, on an Allow too, and never
    /// folded into the tool call's rollup line.
    fn consent_allow(key: &str) -> CapDecisionRecord {
        CapDecisionRecord {
            cap_id: "db:drop".into(),
            key: key.into(),
            action: "request".into(),
            decision: Decision4::Allow,
            mode: "open".into(),
            actor: Actor::Static,
            reason: None,
            rule: None,
            never_rollup: true,
        }
    }

    fn run(f: impl FnOnce()) -> Vec<String> {
        let w = TestWriter::default();
        let sink = w.clone();
        let sub = tracing_subscriber::registry().with(AuditLayer::new(w, Detail::Rollup));
        tracing::subscriber::with_default(sub, f);
        sink.0.lock().unwrap().clone()
    }

    fn issue() -> CredentialIssueRecord {
        CredentialIssueRecord {
            component_ref: "ghcr.io/actpkg/notion@0.1.0".into(),
            session_id: "sess-7".into(),
            key: "notion-work".into(),
            kind: "std:fields".into(),
        }
    }

    #[test]
    fn a_credential_issue_reaches_output_with_no_enclosing_span_at_all() {
        // A record must identify itself from its own fields. Emitting with
        // no span at all is how that is pinned: all four facts have to come
        // off the event, so no later call site can be added that renders an
        // anonymous credential line.
        let out = run(|| emit_credential_issue(&issue()));
        assert_eq!(out.len(), 1, "expected one line, got {out:?}");
        assert!(out[0].contains("notion-work"), "key missing: {}", out[0]);
        assert!(out[0].contains("std:fields"), "kind missing: {}", out[0]);
        assert!(
            out[0].contains("ghcr.io/actpkg/notion@0.1.0"),
            "component missing: {}",
            out[0]
        );
        assert!(out[0].contains("sess-7"), "session missing: {}", out[0]);
    }

    #[test]
    fn a_credential_with_an_empty_kind_is_still_audited() {
        // `kind` comes straight off the stored record and nothing validates
        // it, so `kind: ""` is served to the guest like any other. If the
        // layer keyed on a non-empty value the event would fall through
        // every branch and a secret would cross with no audit line at all.
        let out = run(|| {
            emit_credential_issue(&CredentialIssueRecord {
                kind: String::new(),
                ..issue()
            });
        });
        assert_eq!(out.len(), 1, "expected one line, got {out:?}");
        assert!(out[0].contains("notion-work"), "key missing: {}", out[0]);
        assert!(
            out[0].contains("ghcr.io/actpkg/notion@0.1.0"),
            "component missing: {}",
            out[0]
        );
    }

    #[test]
    fn a_credential_issue_prints_immediately_instead_of_folding_into_the_rollup() {
        // Under Detail::Rollup an allow is folded into a count at span close.
        // A credential issue must not be: "3 credentials issued" does not
        // answer the only question an operator has, which is *which* ones.
        let out = run(|| {
            let span = tool_call_span(&start());
            let _g = span.enter();
            emit_credential_issue(&issue());
            finish_tool_call(&span, Outcome::Ok, Duration::from_millis(1));
        });
        assert_eq!(out.len(), 2, "issue line plus rollup line, got {out:?}");
        assert!(
            out[0].contains("notion-work"),
            "the issue must print before the rollup, got {out:?}"
        );
        assert!(
            !out[1].contains("notion-work"),
            "and must not also be counted in it, got {}",
            out[1]
        );
    }

    #[test]
    fn a_credential_issue_is_not_mistaken_for_a_ceiling_class_or_a_decision() {
        // Both other branches key off `act.capability.id`, which this event
        // does not carry; if the issue branch were removed or ordered after
        // them the event would be silently dropped instead. Emitting all
        // three in one run pins that each still lands in its own shape.
        let out = run(|| {
            let span = instantiation_span("comp", "deadbeef");
            let _g = span.enter();
            emit_ceiling_class(&CeilingClassRecord {
                cap_id: "act:credentials".into(),
                mode: "ask".into(),
                declared: true,
                has_prompt_channel: true,
            });
            emit_credential_issue(&issue());
            emit_cap_decision(&deny());
        });
        let joined = out.join("\n");
        assert!(
            joined.contains("notion-work"),
            "the issue line survived: {joined}"
        );
        assert!(
            joined.contains("act:credentials=ask"),
            "the header still reports the class: {joined}"
        );
        assert!(
            joined.contains("evil.example.com:443"),
            "the denial still rendered: {joined}"
        );
    }

    #[test]
    fn allows_produce_exactly_one_line_at_span_close() {
        let out = run(|| {
            let span = tool_call_span(&start());
            let _g = span.enter();
            for _ in 0..12 {
                emit_cap_decision(&allow("read", "/data/**"));
            }
            finish_tool_call(&span, Outcome::Ok, Duration::from_millis(1400));
        });
        assert_eq!(out.len(), 1, "expected a single rollup line, got {out:?}");
        assert!(out[0].contains("12 read"), "got {}", out[0]);
        assert!(out[0].contains("run_python"));
        // Pins that on_record actually landed the values finish_tool_call
        // recorded, not just that some line got printed.
        assert!(out[0].contains("ok"), "outcome missing, got {}", out[0]);
        assert!(
            out[0].contains("1.4s"),
            "humanised duration missing, got {}",
            out[0]
        );
        // Pins that SpanVisitor actually captured these off the real span
        // (not just that render_rollup can format them when handed a
        // hand-built SpanFields directly, which is all render.rs's own
        // tests exercise).
        assert!(
            out[0].contains("args:9e21c4"),
            "args_sha256 missing, got {}",
            out[0]
        );
        assert!(
            out[0].contains("req:req-1"),
            "request_id missing, got {}",
            out[0]
        );
    }

    #[test]
    fn audit_args_replaces_the_digest_with_full_values_in_the_rollup() {
        // Same fixture-capture concern as the session-id test below: this
        // proves SpanVisitor's TOOL_ARGS arm actually reads the real field
        // off the real span, not just that render_rollup can format an
        // args_json field when handed one directly (render.rs's own tests
        // already cover that in isolation).
        let mut s = start();
        s.args_json = Some(r#"{"path":"/tmp/secret.txt"}"#.into());
        let out = run(|| {
            let span = tool_call_span(&s);
            let _g = span.enter();
            finish_tool_call(&span, Outcome::Ok, Duration::from_millis(5));
        });
        assert_eq!(out.len(), 1, "got {out:?}");
        assert!(
            out[0].contains(r#"args:{"path":"/tmp/secret.txt"}"#),
            "full args missing, got {}",
            out[0]
        );
        assert!(
            !out[0].contains("args:9e21c4"),
            "digest prefix must not also appear once full args are shown, got {}",
            out[0]
        );
    }

    #[test]
    fn a_real_session_id_is_captured_from_the_span_and_rendered() {
        // Every other layer.rs fixture uses session_id: None, so the
        // SESSION_ID capture arm in SpanVisitor::record_str is otherwise
        // never exercised end-to-end: a broken capture and "no session on
        // this call" render identically (no "session:" clause) unless a
        // real, non-empty id is driven through the real span.
        let mut s = start();
        s.session_id = Some("sess-abc123def456".into());
        let out = run(|| {
            let span = tool_call_span(&s);
            let _g = span.enter();
            finish_tool_call(&span, Outcome::Ok, Duration::from_millis(5));
        });
        assert_eq!(out.len(), 1, "got {out:?}");
        assert!(
            out[0].contains("session:sess-abc"),
            "session id missing, got {}",
            out[0]
        );
    }

    #[test]
    fn an_allow_inside_a_non_audit_span_still_folds_into_the_enclosing_tool_call() {
        // SpanState lives only on TARGET_AUDIT spans. A plain span nested
        // between the event and the tool call must not break the fold — the
        // host will instrument exactly this region in a later task.
        let out = run(|| {
            let span = tool_call_span(&start());
            let _g = span.enter();
            {
                let inner = tracing::info_span!("some.other.span");
                let _inner_g = inner.enter();
                emit_cap_decision(&allow("read", "/data/**"));
            }
            finish_tool_call(&span, Outcome::Ok, Duration::from_millis(10));
        });
        assert_eq!(out.len(), 1, "expected a single rollup line, got {out:?}");
        assert!(out[0].contains("1 read"), "got {}", out[0]);
    }

    #[test]
    fn a_consent_allow_never_folds_into_the_rollup_even_inside_a_tool_call() {
        // I2: before this fix, `db:drop=allow` folded into the same rollup a
        // filesystem read does, and a `DROP DATABASE analytics` authorized
        // mid-call rendered as `db:drop: 1 request` — the one fact the line
        // exists to carry (which database) thrown away. `never_rollup: true`
        // must keep it out of the fold and print it the moment it resolves,
        // the same way `render_credential_issue` never folds either.
        let out = run(|| {
            let span = tool_call_span(&start());
            let _g = span.enter();
            emit_cap_decision(&consent_allow("analytics"));
            finish_tool_call(&span, Outcome::Ok, Duration::from_millis(5));
        });
        assert_eq!(
            out.len(),
            2,
            "the consent line plus the rollup line, got {out:?}"
        );
        assert!(
            out[0].contains("analytics"),
            "the consent decision must print immediately and name the key, got {out:?}"
        );
        assert!(
            out[0].contains("db:drop"),
            "and name the class, got {out:?}"
        );
        assert!(
            !out[1].contains("db:drop") && !out[1].contains("analytics"),
            "and must not also be counted in the rollup, got {}",
            out[1]
        );
    }

    #[test]
    fn a_consent_allow_with_nowhere_to_fold_still_prints_once() {
        // Mirrors `an_allow_outside_any_tool_call_still_reaches_the_operator`
        // for the semantic-class case: with no enclosing tool-call span the
        // ordinary fold path is unreachable regardless, but this pins that
        // `never_rollup` does not cause a double-print or a drop here either.
        let out = run(|| emit_cap_decision(&consent_allow("analytics")));
        assert_eq!(out.len(), 1, "got {out:?}");
        assert!(out[0].contains("db:drop"), "got {out:?}");
    }

    #[test]
    fn a_denial_prints_immediately_and_before_the_rollup() {
        let out = run(|| {
            let span = tool_call_span(&start());
            let _g = span.enter();
            emit_cap_decision(&deny());
            emit_cap_decision(&allow("read", "/data/**"));
            finish_tool_call(&span, Outcome::Ok, Duration::from_millis(10));
        });
        assert_eq!(out.len(), 2, "got {out:?}");
        assert!(out[0].contains("deny"), "denial must come first: {out:?}");
        assert!(
            out[1].contains("run_python"),
            "rollup must come last: {out:?}"
        );
    }

    #[test]
    fn full_detail_prints_every_operation() {
        let w = TestWriter::default();
        let sink = w.clone();
        let sub = tracing_subscriber::registry().with(AuditLayer::new(w, Detail::Full));
        tracing::subscriber::with_default(sub, || {
            let span = tool_call_span(&start());
            let _g = span.enter();
            for _ in 0..3 {
                emit_cap_decision(&allow("read", "/data/**"));
            }
            finish_tool_call(&span, Outcome::Ok, Duration::from_millis(5));
        });
        let out = sink.0.lock().unwrap().clone();
        assert_eq!(out.len(), 4, "3 ops + 1 rollup, got {out:?}");
        // Three individual operations, each naming the capability, then the
        // rollup — not e.g. three rollups plus one exception.
        for line in &out[..3] {
            assert!(line.contains("wasi:filesystem"), "got {out:?}");
        }
        assert!(out[3].contains("run_python"), "got {out:?}");
    }

    #[test]
    fn a_decision_outside_any_tool_call_still_reaches_the_operator() {
        // Capability gates can fire during instantiation, before any tool
        // call exists. Those records must not be swallowed.
        let out = run(|| emit_cap_decision(&deny()));
        assert_eq!(out.len(), 1, "got {out:?}");
        assert!(out[0].contains("deny"));
    }

    #[test]
    fn an_allow_outside_any_tool_call_still_reaches_the_operator() {
        // Same scenario as the deny case above, but for an allow: with no
        // enclosing tool-call span there is nowhere to fold it, so it must
        // print immediately rather than being silently dropped.
        let out = run(|| emit_cap_decision(&allow("read", "/data/**")));
        assert_eq!(out.len(), 1, "got {out:?}");
        assert!(out[0].contains("wasi:filesystem"), "got {out:?}");
    }

    #[test]
    fn ask_allow_and_ask_deny_decode_correctly() {
        // `EventVisitor::into_record` matches "ask-allow" / "ask-deny" by
        // string; a typo in either arm would delete every ask outcome from
        // the trail with no test failing. `answered` also attributes both to
        // Actor::User, exercising that decode arm too.
        let out = run(|| {
            emit_cap_decision(&CapDecisionRecord::answered(
                "wasi:filesystem",
                "/data/x",
                true,
                true,
            ));
            emit_cap_decision(&CapDecisionRecord::answered(
                "wasi:http",
                "evil.example.com",
                false,
                true,
            ));
        });
        assert_eq!(out.len(), 2, "got {out:?}");
        assert!(out[0].contains("ask-allow"), "got {out:?}");
        assert!(out[1].contains("ask-deny"), "got {out:?}");
    }

    #[test]
    fn a_quote_bearing_resource_key_survives_rendering_intact() {
        // `%value` fields route through `record_debug`, whose `Debug` output
        // is already the unquoted Display form. A prior `trim_matches('"')`
        // there stripped real leading/trailing quote characters out of
        // guest-controlled data instead of normalising anything.
        let out = run(|| {
            emit_cap_decision(&CapDecisionRecord {
                cap_id: "wasi:filesystem".into(),
                key: "\"payload\".json".into(),
                action: "read".into(),
                decision: Decision4::Deny,
                mode: "ask".into(),
                actor: Actor::Static,
                reason: Some("outside ceiling".into()),
                rule: None,
                never_rollup: false,
            });
        });
        assert_eq!(out.len(), 1, "got {out:?}");
        assert!(
            out[0].contains("\"payload\".json"),
            "quotes must survive intact, got {}",
            out[0]
        );
    }

    #[test]
    fn a_call_that_never_finishes_renders_as_incomplete() {
        // A span created and entered but closed without finish_tool_call
        // ever being called (early return, or dropped) is exactly the case
        // an auditor cares about — it must not render as if outcome "" and
        // duration_ms 0 were a real completed call.
        let out = run(|| {
            let span = tool_call_span(&start());
            let _g = span.enter();
            emit_cap_decision(&allow("read", "/data/**"));
            // Deliberately never call finish_tool_call.
        });
        assert_eq!(out.len(), 1, "got {out:?}");
        assert!(out[0].contains("incomplete"), "got {}", out[0]);
    }

    #[test]
    fn a_panic_while_rendering_does_not_poison_enforcement() {
        // Rendering touches guest-chosen values, so it must run inside the
        // same catch_unwind as the write — not be evaluated before it.
        struct Silent;
        impl AuditWriter for Silent {
            fn write_line(&self, _l: &str) {}
        }
        let layer = AuditLayer::new(Silent, Detail::Rollup);
        layer.emit(|| panic!("render exploded"));
        // Reaching here without unwinding is the assertion.
    }

    #[test]
    fn an_instantiation_span_renders_one_header_line() {
        let out = run(|| {
            let span = instantiation_span("python-eval@0.16.0", "1f3a9c4e");
            let _g = span.enter();
            emit_ceiling_class(&CeilingClassRecord {
                cap_id: "wasi:filesystem".into(),
                mode: "allowlist".into(),
                declared: true,
                has_prompt_channel: true,
            });
            // A declared `ask` class backed by a real prompt channel: not a
            // warning case (see `a_declared_ask_class_with_a_prompt_channel_does_not_warn`)
            // — `has_prompt_channel: true` here is what keeps this test at
            // exactly one line.
            emit_ceiling_class(&CeilingClassRecord {
                cap_id: "wasi:http".into(),
                mode: "ask".into(),
                declared: true,
                has_prompt_channel: true,
            });
        });
        assert_eq!(out.len(), 1, "got {out:?}");
        assert!(
            out[0].contains("wasi:filesystem=allowlist"),
            "got {}",
            out[0]
        );
        assert!(out[0].contains("wasi:http=ask"));
        assert!(out[0].contains("sha256:1f3a9c"));
    }

    #[test]
    fn a_declared_but_denied_class_produces_a_warning_line() {
        let out = run(|| {
            let span = instantiation_span("c@1", "abcdef01");
            let _g = span.enter();
            emit_ceiling_class(&CeilingClassRecord {
                cap_id: "wasi:http".into(),
                mode: "deny".into(),
                declared: true,
                has_prompt_channel: true,
            });
        });
        assert_eq!(out.len(), 2, "header + warning, got {out:?}");
        assert!(out[1].contains("wasi:http"));
        assert!(out[1].contains("not granted"), "got {}", out[1]);
    }

    #[test]
    fn an_undeclared_denied_class_produces_no_warning() {
        // Every class the component never asked for resolves to deny. Warning
        // on those would bury the one signal that matters.
        let out = run(|| {
            let span = instantiation_span("c@1", "abcdef01");
            let _g = span.enter();
            emit_ceiling_class(&CeilingClassRecord {
                cap_id: "wasi:sockets".into(),
                mode: "deny".into(),
                declared: false,
                has_prompt_channel: true,
            });
        });
        assert_eq!(out.len(), 1, "header only, got {out:?}");
    }

    #[test]
    fn a_declared_ask_class_with_no_prompt_channel_produces_a_warning_line() {
        // Headless / ACT-HTTP: `ask` is the configured mode, but there is no
        // channel to ever answer one — every access degrades to deny before
        // a human is asked. The header must still show `ask` unchanged (that
        // is genuinely the configured policy); the warning is what names the
        // real outcome.
        let out = run(|| {
            let span = instantiation_span("c@1", "abcdef01");
            let _g = span.enter();
            emit_ceiling_class(&CeilingClassRecord {
                cap_id: "wasi:http".into(),
                mode: "ask".into(),
                declared: true,
                has_prompt_channel: false,
            });
        });
        assert_eq!(out.len(), 2, "header + warning, got {out:?}");
        assert!(out[0].contains("wasi:http=ask"), "got {}", out[0]);
        assert!(out[1].contains("wasi:http"), "got {}", out[1]);
        assert!(
            out[1].contains("denied"),
            "warning must name the reason, got {}",
            out[1]
        );
    }

    #[test]
    fn a_declared_ask_class_with_a_prompt_channel_does_not_warn() {
        // A TTY or an MCP client offering elicitation means an `ask` really
        // can reach a human — nothing has been refused yet at instantiation
        // time, so this must not warn.
        let out = run(|| {
            let span = instantiation_span("c@1", "abcdef01");
            let _g = span.enter();
            emit_ceiling_class(&CeilingClassRecord {
                cap_id: "wasi:filesystem".into(),
                mode: "ask".into(),
                declared: true,
                has_prompt_channel: true,
            });
        });
        assert_eq!(out.len(), 1, "header only, got {out:?}");
    }

    #[test]
    fn a_writer_that_panics_does_not_poison_enforcement() {
        struct Exploding;
        impl AuditWriter for Exploding {
            fn write_line(&self, _l: &str) {
                panic!("sink exploded");
            }
        }
        let sub = tracing_subscriber::registry().with(AuditLayer::new(Exploding, Detail::Rollup));
        tracing::subscriber::with_default(sub, || {
            emit_cap_decision(&deny());
        });
        // Reaching here without unwinding is the assertion.
    }
}