bugwarden 0.6.0

//! Operator-facing audit event stream: record schema and JSONL file sink.
//!
//! bugwarden places an operator-written guard policy between an MCP client
//! and Bugzilla. On the client side that guard is invisible by design: a
//! hidden bug is indistinguishable from a nonexistent one, filtered search
//! results are dropped without a trace, and the rules doing either are
//! never named. The audit stream is the other half of that bargain — the
//! operator's own record of what was asked and what the guard decided. Its
//! records carry exactly the facts the client must never see (guard
//! verdicts, matched rule names, suppressed bug ids), so the stream goes
//! only to the sinks the operator named: it is never exposed through any
//! MCP surface, and it is never mixed into the diagnostic stderr stream.
//! Diagnostics about the sink itself (a full disk, a failed rotation) go
//! to `tracing` as usual, but carry no event content.
//!
//! A deployment CHOOSES its sinks (revised 2026-08-18, issue #31;
//! [`select_sinks`]): the JSONL file, an OTLP collector ([`AuditExport`],
//! `crate::otel`), both, or none at all. Every configured sink is
//! load-bearing — a failed export puts the sink into failure exactly as a
//! failed write does, and the operator's [`FailMode`] decides what the
//! server then does about tool calls. With both configured the record
//! reaches the file first and the exporter afterwards, so the file never
//! lacks a record the collector has. With neither configured the server
//! serves with no audit trail and no audit gate — a deployment choice,
//! made only by leaving BOTH knobs unset.
//!
//! # File format
//!
//! One JSON object per line (JSONL), schema version 2: see [`AuditEvent`]
//! for the envelope and [`AuditEventKind`] for the record kinds. Records
//! are append-only; [`AuditSink`] optionally rotates the file by size.
//! A failed write can tear the stream: a failed `write(2)` may leave a
//! partial line behind, and the sink heals it by prefixing the next
//! record with a newline. After an outage the file may therefore contain
//! one empty line and possibly one torn (unparsable) line — parsers must
//! skip empty lines and may encounter at most one torn line per outage.
//!
//! # Trust tiers
//!
//! The fields that name a caller, or that let records be joined to one,
//! come from four different places — and reading one at the wrong tier
//! is how an audit trail misleads. The tier is a property of the FIELD,
//! not of the value it happens to hold: a plausible-looking name is
//! still self-declared.
//!
//! | Tier | Fields | What it is worth |
//! |---|---|---|
//! | Self-declared | `client.name`, `client.version` | Whatever the client said about itself. Never verified, by anything: a label, never an identity. |
//! | Server-verified | `client.principal`, `client.work_context` | Only a validator this server trusts may mint one, and nothing self-declared may be promoted into either. Reserved: no verifier exists, so neither appears in a record THIS BUILD writes — a later build that can prove a caller fills them without moving `v`. |
//! | Server-minted | `session.id` | Within one deployment's stream, proves that records sharing it passed through one transport session — and, where that session began with a handshake, joins them to its `initialize`. Says nothing about WHO was on the other end. |
//! | Correlation hint | `trace`, `request.id`, `session.remote` | Client-chosen or path-dependent — a `traceparent` any caller may stamp, an id the client picked, a peer address that is the last proxy's behind one. Useful for joining records; evidence of nothing. |
//!
//! This table is normative and versioned with the schema: a new field
//! that names or correlates a caller belongs to one of these four tiers,
//! and adding one without saying which is the omission that lets a
//! reader promote a claim into a fact. The REST of a record — verdicts,
//! rule names, counts, timings, the tool called — is the server's own
//! account of what it did rather than a claim about anybody, so the
//! tiering question does not arise for it. TWO EXCEPTIONS, and the first
//! is the case a future field is likeliest to hit: [`RequestInfo::params`]
//! is CLIENT-authored content wherever the allowlist passes it verbatim —
//! the `_len` and `_overlong_keys` reductions beside it are the server's
//! (both names reserved at every depth, #241) — not the server's account
//! of anything, so a value read out of it is a correlation hint at best
//! and never an identity — a grouping handle a client was given and
//! handed back included. The second: [`RequestInfo::tool`] is that
//! account only while the name is one this server serves, since a call
//! it will not serve is recorded too — an unknown name is the client's
//! own text, capped for that reason (#243), and groups records by
//! nothing but itself.
//!
//! # What can never appear in a record
//!
//! The schema is closed on purpose: every struct rejects unknown fields,
//! every enum is a fixed vocabulary, and no dedicated field exists for
//! the Bugzilla API key, incoming request headers, or free-text bug
//! content (summaries, comments, attachments). The one open field is
//! [`RequestInfo::params`]: it is fed exclusively through an allowlist
//! of client-authored parameter keys at the recording call site, and
//! data fetched from Bugzilla never passes through it.
//! [`AuditError`] follows the same rule — it names the failed
//! operation and carries the underlying I/O error, never the record that
//! failed to be written, so it stays safe to surface in diagnostics.
//!
//! Every client-chosen string a record does carry is bounded at the
//! recording call site: allowlisted `params` values and over-cap key
//! prefixes (#211, #220), the self-declared identity (#191),
//! [`RequestInfo::tool`] — client text whenever it names a tool this
//! server does not serve (#243) — and [`RequestInfo::id`] (#248) are
//! capped at 1024 characters; `trace` is parsed to fixed-width hex ids
//! instead.
//!
//! # Ordering, durability, and blocking
//!
//! [`AuditSink::record`] is deliberately synchronous: when it returns
//! `Ok`, the record has been handed to the kernel (and, with
//! [`AuditConfig::fsync`], flushed to stable storage), so a caller can
//! guarantee "record persisted before the response is returned". All
//! writes serialize through one mutex; `seq` is assigned under that lock,
//! so file order equals `seq` order — a documented guarantee. Async
//! callers must wrap calls in `tokio::task::block_in_place` or
//! `tokio::task::spawn_blocking` (this module itself has no tokio
//! dependency).
//!
//! One deliberate asymmetry in the loss accounting: with
//! [`AuditConfig::fsync`] enabled, a failed `sync_data` after a
//! successful `write(2)` counts the record as dropped even though its
//! bytes may already be in the file. The same `seq` can then appear
//! twice on disk, and the following `audit_gap` may over-report the
//! loss. The direction is chosen: the accounting may over-report,
//! never under-report.
//!
//! [`AuditConfig::fsync`]: crate::audit::AuditConfig::fsync
//! [`AuditError`]: crate::audit::AuditError
//! [`AuditEvent`]: crate::audit::AuditEvent
//! [`AuditEventKind`]: crate::audit::AuditEventKind
//! [`AuditExport`]: crate::audit::AuditExport
//! [`AuditSink`]: crate::audit::AuditSink
//! [`AuditSink::record`]: crate::audit::AuditSink::record
//! [`FailMode`]: crate::audit::FailMode
//! [`RequestInfo::params`]: crate::audit::RequestInfo::params
//! [`select_sinks`]: crate::audit::select_sinks

use std::collections::BTreeMap;
use std::fs::{File, OpenOptions};
use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::{Duration, Instant};

use anyhow::Context as _;
use serde::{Deserialize, Serialize};
use sha2::{Digest as _, Sha256};

/// The audit record schema version stamped into every event (`v`).
///
/// It tracks the STRUCTURE a reader must handle — the fields, their
/// names, their order and their types, and the record kinds — and not
/// the meaning of what those fields hold. A field whose definition is
/// corrected without changing the wire shape keeps `v`, so a corpus at
/// one version can span both readings, with the deployed BUILD and not
/// this number as the boundary between them. What that structure may
/// still GROW inside a version is the rule below; the rest of it is
/// frozen for the version's life.
///
/// # The growth rule
///
/// Within one version, two changes are allowed:
///
/// - adding an OPTIONAL field that is absent when unset;
/// - adding a new RECORD KIND.
///
/// Neither MAY change how an existing field is read. That is a
/// requirement of the rule, not an observation about it: an optional
/// field that redefines a field already shipping — a hypothetical
/// `guard.counts_are_floors`, absent when unset — is structurally
/// additive and still forbidden, because it is #68's injury again, a
/// meaning moving invisibly under an unchanged `v`. Such a change moves
/// the number however optional it looks.
///
/// Past that the two differ, and the difference is not symmetric: a new
/// KIND never reaches a consumer selecting on `event` — one filtering
/// `event == "tool_call"` does not see it — while a new FIELD has no
/// such escape and lands in the very records that consumer already
/// selects. What they share is
/// the direction of the break they can cause: `deny_unknown_fields`
/// means a reader built BEFORE the addition rejects a line carrying it,
/// so staying within a version buys old files new readers, never old
/// readers new files. The goldens block in this module's tests pins the
/// discipline that goes with it.
///
/// REMOVING, RENAMING, REORDERING or RETYPING a field moves this number.
/// So a new kind is not a bump, and #32's refusal record — should it
/// ever be recorded rather than logged — would not force one.
///
/// Widening a CLOSED VOCABULARY ([`GapReason`], [`Verdict`],
/// [`OutcomeClass`], [`TransportKind`]) is NOT the additive case, and it
/// does move the number. An added field or kind is something a reader
/// can decline to look at; a new VARIANT lands inside a field every
/// reader of that kind already reads, and there is nothing to decline.
/// That missing escape is the difference, so a vocabulary change does
/// not inherit the licence the other two have.
///
/// # What v2 changed
///
/// - `guard.suppressed_count` is GONE, split into
///   [`GuardInfo::suppressed_ids_count`] and
///   [`GuardInfo::suppressed_other_count`] (issue #68). No total ships:
///   a total beside its own parts can disagree with them, and losing the
///   shared field name is what makes the break loud rather than silent.
/// - [`ClientInfo::work_context`] is reserved beside
///   [`ClientInfo::principal`], and both are documented as CONTRACTS
///   about who may fill them rather than as the `None` they hold today
///   (issue #34).
///
/// v1 carried a correction the split supersedes: `suppressed_count`
/// became the SUM of the two tallies where it had been their maximum,
/// invisible on the wire and therefore undetectable from a record. That
/// invisibility is precisely the argument for making the distinction
/// structural now.
///
/// # The v1 break, and its exact scope
///
/// Deliberate, and load-bearing. A v2 reader REJECTS every v1
/// `tool_call` line that carries a `guard` object:
/// `deny_unknown_fields` refuses `suppressed_count`, and a reader that
/// accepted it would have to invent a split for a number that cannot be
/// split after the fact.
///
/// It rejects nothing else. v1 `initialize`, `audit_gap`, and
/// guard-less `tool_call` lines still parse — and parse carrying
/// `v: 1`, because this reader does not dispatch on `v`. A consumer
/// holding a mixed corpus must therefore split on that field itself
/// before assuming either version's semantics.
pub const SCHEMA_VERSION: u32 = 2;

/// How often a persistently failing sink repeats its `tracing` diagnostic.
const FAILURE_LOG_INTERVAL: Duration = Duration::from_secs(60);

/// Upper bound for [`AuditConfig::rotate_keep`]: rotation shifts every
/// kept file with one rename syscall each while all audit writes wait on
/// the sink lock, so the kept window must stay small.
const MAX_ROTATE_KEEP: u32 = 10_000;

// ---------------------------------------------------------------------------
// Configuration
// ---------------------------------------------------------------------------

/// What the server does with a tool call whose audit record cannot be
/// persisted.
///
/// This module only parses and stores the choice; enforcement lives with
/// the code that wires the sink into the request path.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FailMode {
    /// Keep serving. The gap is still visible in the stream afterwards
    /// (see [`AuditGapEvent`]), but requests made while the sink is down
    /// are lost to the record. Availability over accountability.
    Open,
    /// Reads the guard fully allows still proceed unaudited; write
    /// operations and any call where the guard suppressed, denied, or
    /// refused something are refused until their record can be persisted
    /// again.
    ClosedWritesDenials,
    /// Every tool call is refused until its record can be persisted.
    /// Accountability over availability: an unmonitored full disk turns
    /// the server off.
    ClosedAll,
}

fn default_rotate_max_bytes() -> u64 {
    64 * 1024 * 1024
}

fn default_rotate_keep() -> u32 {
    8
}

fn default_suppressed_ids() -> bool {
    true
}

/// Audit sink configuration, loaded from an operator-owned TOML file.
///
/// Parsing is strict: unknown keys anywhere are an error at load time, so
/// a typo cannot silently disable a setting. `examples/audit.toml` in the
/// repository is a fully commented worked example.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AuditConfig {
    /// The JSONL audit file. Still REQUIRED in a configuration file —
    /// this is `Option` only so a deployment that has no configuration
    /// file can construct a fileless sink ([`AuditConfig::fileless`]),
    /// and a TOML document that omits `path` is a load error as before.
    /// The parent directory is created (mode 0700) if missing; the file
    /// itself is created mode 0600 and only ever appended to.
    pub path: Option<PathBuf>,
    /// `sync_data()` after every record. The default `false` still
    /// survives a killed process — `write(2)` puts the record in the page
    /// cache before [`AuditSink::record`] returns — but only `true`
    /// additionally survives power loss, at a per-record latency cost.
    #[serde(default)]
    pub fsync: bool,
    /// What to do with tool calls while records cannot be persisted.
    /// When absent, the server derives the mode from the transport at
    /// wiring time: `stdio` (a local, single-operator session) defaults
    /// to `open`, `http` (remote clients) defaults to `closed_all`.
    #[serde(default)]
    pub fail_mode: Option<FailMode>,
    /// Rotate the live file before a record would push it past this many
    /// bytes. `0` disables built-in rotation entirely — for operators who
    /// run external rotation instead. Default: 67108864 (64 MiB).
    #[serde(default = "default_rotate_max_bytes")]
    pub rotate_max_bytes: u64,
    /// How many rotated files (`<path>.1` … `<path>.N`) are kept; older
    /// ones are deleted at rotation time. Must be at least 1 while
    /// `rotate_max_bytes` is non-zero, and at most 10000 (both validated
    /// at load). Shrinking it between runs strands existing
    /// higher-numbered `<path>.N` files: bugwarden never deletes them —
    /// remove them manually. Default: 8.
    #[serde(default = "default_rotate_keep")]
    pub rotate_keep: u32,
    /// Whether records may carry the ids of guard-suppressed bugs
    /// ([`GuardInfo::suppressed_ids`]). When `false`, records carry only
    /// counts. This module stores the flag; the recording call sites
    /// enforce it when building [`GuardInfo`]. Default: `true`.
    #[serde(default = "default_suppressed_ids")]
    pub suppressed_ids: bool,
}

impl AuditConfig {
    /// The configuration of a deployment that keeps no audit file, for
    /// `BUGWARDEN_AUDIT_CONFIG=none`.
    ///
    /// There is no document to carry the other knobs, so they take their
    /// defaults and [`AuditConfig::fail_mode`] is left for the caller to
    /// resolve. Such a deployment MUST have another sink; `main` enforces
    /// that before it gets here.
    #[must_use]
    pub fn fileless() -> AuditConfig {
        AuditConfig {
            path: None,
            fsync: false,
            fail_mode: None,
            rotate_max_bytes: default_rotate_max_bytes(),
            rotate_keep: default_rotate_keep(),
            suppressed_ids: default_suppressed_ids(),
        }
    }

    /// Strict parse + validation of an audit configuration document.
    ///
    /// Rejects unknown keys everywhere (`deny_unknown_fields`), then
    /// enforces that `rotate_keep >= 1` whenever built-in rotation is
    /// enabled (`rotate_max_bytes > 0`) and that `rotate_keep` never
    /// exceeds 10000.
    pub fn from_toml_str(s: &str) -> anyhow::Result<AuditConfig> {
        let cfg: AuditConfig =
            toml::from_str(s).context("failed to parse audit configuration TOML")?;
        // `path` is `Option` only for the fileless constructor; a DOCUMENT
        // without one is a load error as it always was — the audit file is
        // turned off by BUGWARDEN_AUDIT_CONFIG=none, never by omission.
        if cfg.path.is_none() {
            anyhow::bail!("audit configuration names no `path`");
        }
        cfg.validate()?;
        Ok(cfg)
    }

    /// Read + [`AuditConfig::from_toml_str`], with the file path in every
    /// error.
    pub fn load(path: &Path) -> anyhow::Result<AuditConfig> {
        let s = std::fs::read_to_string(path).with_context(|| {
            format!("failed to read audit configuration file {}", path.display())
        })?;
        Self::from_toml_str(&s)
            .with_context(|| format!("invalid audit configuration file {}", path.display()))
    }

    fn validate(&self) -> anyhow::Result<()> {
        if self.rotate_max_bytes > 0 && self.rotate_keep == 0 {
            anyhow::bail!(
                "rotate_keep must be at least 1 while rotate_max_bytes is non-zero: \
                 rotation renames the live file to <path>.1, and keeping zero rotated \
                 files would delete the records it just wrote; set rotate_max_bytes = 0 \
                 instead to hand rotation to an external tool"
            );
        }
        if self.rotate_keep > MAX_ROTATE_KEEP {
            anyhow::bail!(
                "rotate_keep must be at most {MAX_ROTATE_KEEP}: every rotation shifts \
                 each kept file with its own rename syscall while all audit writes \
                 wait, so a larger window would stall the server; raise \
                 rotate_max_bytes instead to retain more history"
            );
        }
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Sink selection
// ---------------------------------------------------------------------------

/// The `--audit-config` / `BUGWARDEN_AUDIT_CONFIG` value that disables
/// the audit FILE by name. Compared by exact bytes, like a rule name: a
/// real file that happens to be called `none` is reachable as `./none`.
pub const AUDIT_CONFIG_NONE: &str = "none";

/// Which audit sinks this deployment runs (issue #31, revised 2026-08-18).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SinkSelection {
    /// No sink at all: the server serves normally with NO audit trail and
    /// no audit gate. Expressible only by leaving both knobs unset.
    NoAudit,
    /// The JSONL file alone — every pre-#31 deployment, unchanged.
    FileOnly,
    /// File and OTLP: the write precedes the export, and either sink
    /// failing gates serving under the one [`FailMode`].
    Both,
    /// OTLP alone, on a fileless sink (`BUGWARDEN_AUDIT_CONFIG=none`).
    OtlpOnly,
}

/// Resolve the two knobs — the audit-config path and whether an OTLP
/// endpoint is configured — into the deployment's sink selection.
///
/// The rule that shapes the matrix: turning the FILE off is an explicit
/// act (`BUGWARDEN_AUDIT_CONFIG=none`), never an inference from absence.
/// A typo that unsets one variable must fail the start, not silently
/// reshape the audit trail — so the two cells where absence and an
/// explicit choice could be confused are startup errors:
///
/// - An OTLP endpoint with NO audit-config value could mean "OTLP-only"
///   or "I mistyped the file configuration"; the server refuses to guess.
/// - `none` with no OTLP endpoint would leave zero sinks on an explicit
///   request for one; running without an audit trail is legitimate but is
///   asked for by leaving both knobs unset, not by this.
///
/// # Errors
///
/// The two ambiguous cells above, each naming the variables and the two
/// ways to resolve it.
pub fn select_sinks(
    audit_config: Option<&Path>,
    otlp_configured: bool,
) -> anyhow::Result<SinkSelection> {
    let file = match audit_config {
        None => None,
        Some(path) if path.as_os_str() == AUDIT_CONFIG_NONE => Some(false),
        Some(_) => Some(true),
    };
    match (file, otlp_configured) {
        (None, false) => Ok(SinkSelection::NoAudit),
        (Some(true), false) => Ok(SinkSelection::FileOnly),
        (Some(true), true) => Ok(SinkSelection::Both),
        (Some(false), true) => Ok(SinkSelection::OtlpOnly),
        (Some(false), false) => anyhow::bail!(
            "--audit-config/BUGWARDEN_AUDIT_CONFIG is `{AUDIT_CONFIG_NONE}` but no OTLP \
             endpoint is configured, which would leave no audit sink at all; set \
             OTEL_EXPORTER_OTLP_ENDPOINT for OTLP-only auditing, or unset \
             BUGWARDEN_AUDIT_CONFIG entirely if this deployment is meant to run \
             without an audit trail"
        ),
        (None, true) => anyhow::bail!(
            "OTEL_EXPORTER_OTLP_ENDPOINT is set, so audit records are exported and \
             their delivery gates serving — but no audit file is configured; point \
             --audit-config/BUGWARDEN_AUDIT_CONFIG at an audit configuration to run \
             file and OTLP together, or set it to `{AUDIT_CONFIG_NONE}` for OTLP-only \
             auditing"
        ),
    }
}

// ---------------------------------------------------------------------------
// Event schema (v2)
// ---------------------------------------------------------------------------

/// One audit record: the envelope common to every kind.
///
/// Serialized as a single JSON object (one line of the JSONL file) with
/// the envelope fields first, then the fields of the [`AuditEventKind`]
/// flattened alongside them under an `event` tag.
///
/// This is the only struct in the schema without `deny_unknown_fields`:
/// serde cannot combine that attribute with `flatten`. Unknown top-level
/// keys are still rejected — they flow into the flattened kind, whose
/// structs are closed.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AuditEvent {
    /// Schema version. Write side: this build stamps [`SCHEMA_VERSION`]
    /// into every record it produces. READ side is not the same claim —
    /// the deserializer neither checks this field nor dispatches on it,
    /// so a line off disk carries whatever version wrote it, and a v1
    /// line outside the v2 break parses with `v: 1` intact (see
    /// [`SCHEMA_VERSION`]). A consumer holding a mixed corpus must split
    /// on this value before assuming either version's semantics.
    pub v: u32,
    /// Wall-clock record time, RFC 3339 UTC with millisecond precision.
    /// Stamped by the sink. `seq`, not `ts`, is the ordering authority —
    /// the wall clock can step backwards.
    pub ts: String,
    /// Per-process monotonic sequence number, assigned by the sink under
    /// its write lock, starting at 1. File order equals `seq` order, and
    /// only records that reached the file consume a number, so the seqs
    /// on disk are contiguous — with one exception: under
    /// [`AuditConfig::fsync`], a failed sync after a successful write
    /// counts the record as dropped even though its bytes may be on
    /// disk, so a seq can appear twice and the next `audit_gap` may
    /// over-report (never under-report) the loss. A restarted process
    /// starts again at 1; across restarts, ordering comes from file
    /// order and `ts`.
    pub seq: u64,
    /// The MCP session this record belongs to.
    pub session: SessionInfo,
    /// The record kind and its payload, flattened into the envelope.
    #[serde(flatten)]
    pub kind: AuditEventKind,
}

/// The audit record kinds, tagged `event` in the serialized form.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "event", rename_all = "snake_case")]
pub enum AuditEventKind {
    /// A tool call reached the server (whatever the guard then decided).
    /// Boxed: this payload dwarfs the other variants, and boxing keeps the
    /// enum small on the stack; the serialized form is unaffected.
    ToolCall(Box<ToolCallEvent>),
    /// An MCP session performed the `initialize` handshake.
    Initialize(InitializeEvent),
    /// One or more records could not be persisted; see [`AuditGapEvent`].
    AuditGap(AuditGapEvent),
}

/// Payload of a `tool_call` record: one MCP tool invocation, from request
/// to outcome, including what the guard decided about it.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ToolCallEvent {
    /// The calling client as it introduced itself.
    pub client: ClientInfo,
    /// Distributed-trace correlation ids, when the transport carried any.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub trace: Option<TraceContext>,
    /// The request as the client made it (allowlisted parameters only).
    pub request: RequestInfo,
    /// What the guard decided. Absent for tools that perform no guard
    /// assessment (`bug_url` computes a URL locally and contacts nothing).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub guard: Option<GuardInfo>,
    /// The Bugzilla side of the call. Absent when Bugzilla was never
    /// contacted: a local tool, a refusal answered from the request alone,
    /// or the pre-dispatch gate. Absence means zero requests, never
    /// "not measured".
    #[serde(skip_serializing_if = "Option::is_none")]
    pub upstream: Option<UpstreamInfo>,
    /// How the call ended, from the client's point of view.
    pub outcome: OutcomeInfo,
}

/// Payload of an `initialize` record: a client sent an `initialize`.
///
/// Usually that opened a session, and [`SessionInfo::id`] then joins this
/// anchor to the records it anchors. A `2026-07-28` `initialize` over http
/// is the exception: rmcp routes it down the handshake-free path, so it is
/// answered and recorded but opens nothing, and the record carries no id.
/// It is written anyway: recording every handshake unconditionally is the
/// invariant, and joins are on id equality, so an anchor with no id anchors
/// nothing and gathers nothing that belongs elsewhere either. Over stdio the
/// same request is an ordinary handshake and does open a session.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct InitializeEvent {
    /// The client as it introduced itself in the handshake — the
    /// REQUEST's own `clientInfo`, never a peer identity the transport
    /// synthesised.
    pub client: ClientInfo,
    /// The MCP protocol version the session negotiated — the revision
    /// actually spoken, which is not always the one the client asked for:
    /// a client requesting a revision this build does not serve is handed
    /// the server default, and this field records what it got.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub protocol_version: Option<String>,
}

/// Payload of an `audit_gap` record: the sink failed to persist records
/// and has recovered. Written as the first record after the first
/// successful write, so a gap in the stream is always visible in the
/// stream itself.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AuditGapEvent {
    /// How many records were lost while the sink was failing.
    pub dropped: u64,
    /// Why the most recent of them was lost. A closed vocabulary on
    /// purpose: a free-text reason would be a hole in the schema's
    /// no-secrets guarantee.
    pub reason: GapReason,
}

/// Why records were lost. See [`AuditGapEvent`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum GapReason {
    /// Serializing or writing a record failed.
    WriteError,
    /// Rotating the file failed.
    RotationError,
}

/// The MCP session a record belongs to.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SessionInfo {
    /// Server-MINTED session identifier, present when the transport
    /// actually opened a session: over http the id rmcp minted, over
    /// stdio the process itself. Never a client-supplied value — the
    /// `mcp-session-id` request header is not read (#180).
    ///
    /// What it GUARANTEES is grouping WITHIN ONE DEPLOYMENT'S STREAM:
    /// records sharing an id passed through one transport session. Where
    /// that session began with a handshake, its `initialize` record carries
    /// the same id, so a tool record joins the handshake that anchors it.
    ///
    /// The anchor is not guaranteed, and the gap is not the http one. A
    /// 2026-07-28 client over STDIO may open with an ordinary request and
    /// never send `initialize` at all — rmcp serves that first message and
    /// keeps the connection — while this id, minted from the process, is
    /// stamped on every record regardless. A whole session of records can
    /// therefore share an id no `initialize` names. They stay truthful: the
    /// id is unforgeable and no other session shares it, so an unanchored
    /// group is never joined to the WRONG session — it simply has nothing to
    /// join to.
    ///
    /// That scope is load-bearing, not boilerplate. Over stdio the id is
    /// `<pid>-<startup epoch seconds>`, unique on a host and not beyond
    /// it, and no field of a record names the deployment that wrote it.
    /// A collector merging several deployments must therefore key on the
    /// exporter's `service.name` beside the id — and that name defaults
    /// to the constant `bugwarden`, so an operator aggregating more than
    /// one deployment has to set it.
    ///
    /// What it does NOT: it is not an identity. A session is a
    /// connection, and nothing about who was on the other end of it
    /// follows from the id. Nor does it exist on the handshake-free
    /// per-request path OVER HTTP, which opens no session — there the
    /// only candidate is the `mcp-session-id` request header, unvalidated
    /// client text that any caller could stamp with a live session's id,
    /// and a forgeable id is worse than none. Grouping there is
    /// [`SessionInfo::remote`] plus nothing. Over stdio the same
    /// lifecycle keeps its id: that one is minted from the process, not
    /// read off a request, so there is nothing to forge.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub id: Option<String>,
    /// Which transport carried the session.
    pub transport: TransportKind,
    /// Remote peer address for network transports, absent for stdio.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub remote: Option<String>,
}

/// The transport a session arrived over.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TransportKind {
    /// Local stdin/stdout session.
    Stdio,
    /// Streamable-HTTP session.
    Http,
}

/// The calling client's identity, as far as one exists.
///
/// Two opposite contracts in one struct. [`ClientInfo::name`] and
/// [`ClientInfo::version`] are SELF-DECLARED — whatever the client said
/// about itself, in the `initialize` handshake (the only source this
/// build has) or, on a handshake-free revision, in the request's own
/// `_meta`. The source does not change the trust level; neither value is
/// ever verified against anything. [`ClientInfo::principal`] and
/// [`ClientInfo::work_context`] are the other contract: SERVER-VERIFIED,
/// absent until a verifier exists, and never filled from anything
/// self-declared.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ClientInfo {
    /// Client name as the client declared it: the `initialize`
    /// handshake, or the request's own `_meta` on the handshake-free
    /// 2026-07-28 lifecycle this build serves — the same trust level
    /// either way. Self-declared and therefore untrusted: treat it as a
    /// label, never as an identity. Absent when the caller declared
    /// nothing, which the per-request lifecycle permits (`clientInfo` is
    /// optional there) — absent is the honest answer, and never the
    /// SDK's own build identity, which is what the transport synthesises
    /// for a peer that never handshook.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    /// Client version, from the same source and at the same trust level
    /// as [`ClientInfo::name`].
    #[serde(skip_serializing_if = "Option::is_none")]
    pub version: Option<String>,
    /// SERVER-VERIFIED caller identity: who called, as this deployment
    /// PROVED it — not who the caller says it is.
    ///
    /// Reserved. No verifier exists, so this field is absent from every
    /// record THIS BUILD writes — which is a statement about the build,
    /// not about the version: the slot is reserved precisely so that a
    /// build which can prove a caller starts filling it without moving
    /// `v`. Only a validator this server trusts may ever mint a value
    /// here — the per-caller credential direction in issue #32 — and
    /// nothing self-declared may be promoted into it, not
    /// [`ClientInfo::name`] and not anything a request carries in
    /// `_meta`. The deployment-wide bearer token does not qualify
    /// either: it authenticates a deployment, not a caller.
    ///
    /// Not the same fact as [`ClientInfo::work_context`] — this one is
    /// WHO.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub principal: Option<String>,
    /// SERVER-VERIFIED work item the call was authorized under: a
    /// maintenance-update identifier, a ticket, whatever a deployment's
    /// validator binds its credential to. The format is deliberately not
    /// pinned here — a syntax written into this contract is one issue
    /// #32's validator would owe.
    ///
    /// Reserved, under [`ClientInfo::principal`]'s rule: only a trusted
    /// validator may fill it, nothing self-declared may be promoted into
    /// it, and it is absent from every record this build writes — again a
    /// statement about the build, since filling it is a value change and
    /// not a version change.
    ///
    /// Optional even once verification exists — a fleet-level agent
    /// belongs to no work item. So absence carries two readings across
    /// the life of v2: "no verifier yet" while this remains reserved,
    /// "this caller belongs to no work item" once one ships. Like #68's
    /// correction inside v1, the boundary is the deployed BUILD and not
    /// `v`, which tracks structure and not meaning.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub work_context: Option<String>,
}

/// W3C-style trace correlation ids, when the transport carried any.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TraceContext {
    /// Trace id (hex, as received).
    pub trace_id: String,
    /// Span id (hex, as received).
    pub span_id: String,
}

impl TraceContext {
    /// Parse a W3C `traceparent` header value (version 00 only) into
    /// trace ids.
    ///
    /// Strict by design: anything not exactly
    /// `00-<32 lowercase hex>-<16 lowercase hex>-<2 lowercase hex>`
    /// (55 bytes) with non-zero trace and span ids is `None`. In
    /// particular:
    ///
    /// - The length check comes FIRST, on bytes — a cheap rejection of
    ///   oversized input, and no later step slices a string, so
    ///   multibyte unicode can never panic the parser.
    /// - The version must be exactly `00`. That rejects `ff` (forbidden
    ///   by W3C) AND every future version: W3C allows lenient forward
    ///   parsing, but this parser refuses to store ids from a format
    ///   revision it cannot fully validate.
    /// - Hex means ASCII lowercase `[0-9a-f]` only; uppercase is invalid
    ///   per W3C.
    /// - An all-zero trace id or span id is invalid per W3C.
    /// - The flags byte pair is validated as hex but not stored — the
    ///   schema has no field for it.
    ///
    /// The input is client-controlled bytes headed for operator logs, so
    /// it is never logged here, valid or not: rejection is silent
    /// (I12-adjacent hygiene).
    pub(crate) fn from_traceparent(value: &str) -> Option<TraceContext> {
        fn lower_hex(bytes: &[u8]) -> bool {
            bytes
                .iter()
                .all(|b| b.is_ascii_digit() || matches!(b, b'a'..=b'f'))
        }
        let bytes = value.as_bytes();
        if bytes.len() != 55 {
            return None;
        }
        // Fixed layout: bytes 0-1 version, byte 2 `-`, bytes 3-34
        // trace-id, byte 35 `-`, bytes 36-51 span-id, byte 52 `-`,
        // bytes 53-54 flags.
        if &bytes[0..2] != b"00" {
            return None;
        }
        if bytes[2] != b'-' || bytes[35] != b'-' || bytes[52] != b'-' {
            return None;
        }
        let trace_id = &bytes[3..35];
        let span_id = &bytes[36..52];
        let flags = &bytes[53..55];
        if !lower_hex(trace_id) || !lower_hex(span_id) || !lower_hex(flags) {
            return None;
        }
        if trace_id.iter().all(|&b| b == b'0') || span_id.iter().all(|&b| b == b'0') {
            return None;
        }
        // Both ranges are validated ASCII hex, so the conversions cannot
        // fail; going through `from_utf8` keeps the parser panic-free by
        // construction.
        Some(TraceContext {
            trace_id: std::str::from_utf8(trace_id).ok()?.to_owned(),
            span_id: std::str::from_utf8(span_id).ok()?.to_owned(),
        })
    }
}

/// The request as the client made it.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RequestInfo {
    /// MCP tool name, capped at 1024 characters by the recording call
    /// site: a call the router will not serve is recorded before it is
    /// refused, so this is CLIENT-authored text whenever it is not one
    /// of the served names (#243).
    pub tool: String,
    /// JSON-RPC request id, when the transport exposes one.
    ///
    /// CLIENT-chosen, so a correlation hint and never an identity:
    /// nothing stops two callers picking the same id, or one caller
    /// reusing one.
    ///
    /// A JSON-RPC string id is client text of any length, so the
    /// recording call site caps it at 1024 characters; the reply still
    /// echoes the full id, since the cap is on the record and never on
    /// the wire (#248).
    ///
    /// Records are per-ATTEMPT, not per intended call. A client that
    /// loses a response and re-issues the call does so under a new id,
    /// and the re-issued call performs its upstream work again — so two
    /// records is the true account of what reached Bugzilla.
    /// Deduplicating them would under-report real mutations, the one
    /// direction this stream must not err in.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub id: Option<String>,
    /// CLIENT-authored request parameters, passed through an allowlist
    /// by the recording call site: identifiers, projections, switches —
    /// parameters whose values the client chose. Verbatim only where
    /// allowlisted: the `_len` and `_overlong_keys` reductions are
    /// server-written, both names reserved at every depth so a client
    /// key spelling either is relocated, never passed through (#241).
    /// Bug content fetched from Bugzilla must never be placed here; a
    /// tool result is not a parameter.
    pub params: BTreeMap<String, serde_json::Value>,
}

/// What the guard decided about a tool call.
///
/// Everything in here is server-side-only fact: verdicts, rule names, and
/// suppression counts are precisely what the guard withholds from the
/// client, and they reach the operator only through this stream.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct GuardInfo {
    /// The overall verdict for the call.
    pub verdict: Verdict,
    /// Name of the rule that decided the verdict, when a single rule did.
    ///
    /// A call the policy DEFAULT decided records the literal `"default"` —
    /// NOT an absent field. Beside the operator's own rule names the guard
    /// emits four synthetic ones: `"default"` (no rule matched),
    /// `"min_bug_age_days"` (the age quarantine denied before any rule
    /// ran), `"<rule>:unreadable-metadata"` (a granting rule whose verdict
    /// hinged on metadata nobody could read, I4), and `"unavailable"` (the
    /// classification fetch never reached the bug). They share one
    /// namespace with the operator's names, and for a policy loaded from
    /// TOML validation reserves all four against it — such a policy
    /// defining a rule literally called `default` is a startup error — so
    /// this field names what decided and proves which kind of thing it was.
    ///
    /// Absent only where no single rule decided the call:
    ///
    /// - a refusal answered from the request alone — bad arguments, too
    ///   many ids;
    /// - the pre-dispatch audit gate, where the guard never ran at all;
    /// - a search, whose verdict is the window's and not one bug's;
    /// - the create gate, refusing or granting: it judges the request as a
    ///   whole and names no rule either way;
    /// - an id the guard produced no matching [`Access`] for — none at
    ///   all, or one whose variant the re-checked outcome contradicts;
    /// - withheld attachment content after a granted assessment, and the
    ///   constant-cost padding assessment against bug id 0, neither of
    ///   which any rule decided.
    ///
    /// A tool the router never carried is a different case again, not an
    /// absent rule: read-only mode, `global.disabled_tools` and discovery
    /// being off REMOVE the route at construction (I13), so the call
    /// reaches no guard, touches no cell, and its record carries no
    /// `guard` object at all.
    ///
    /// [`AuditCell`]'s worst-wins merge carries this through: a rule-less
    /// note that outranks a rule-decided one clears the rule. The
    /// client-requested windows are the one exception — they decide
    /// nothing, so they keep the granting rule
    /// ([`AuditCell::note_redacted_client_window`]).
    ///
    /// Recording `"default"` rather than absence is deliberate, and as
    /// of schema v2 it is DECIDED rather than deferred (issue #67):
    /// absence already means "no single rule decided" — the enumerated
    /// cases above — so encoding the default as absence would load a
    /// second meaning onto it that no reader could separate from the
    /// first. Recorded in DESIGN.md under "Audit stream".
    ///
    /// [`Access`]: bugwarden_core::policy::Access
    #[serde(skip_serializing_if = "Option::is_none")]
    pub rule: Option<String>,
    /// Digest of the policy document in force, so a record can be tied to
    /// the exact policy that produced it.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub policy_hash: Option<String>,
    /// How many withheld items had a bug id of their own: the size of
    /// the id set behind [`GuardInfo::suppressed_ids`] — I14-scrubbed
    /// links, scrubbed duplicate-marker ids, search rows the scan
    /// dropped by verdict.
    ///
    /// Counted from that set BEFORE [`AuditConfig::suppressed_ids`] can
    /// elide the list, so it survives the switch: with ids off this is
    /// the only surviving statement of how many bugs were withheld, and
    /// it must never be re-derived from the emitted vector.
    pub suppressed_ids_count: u64,
    /// How many withheld items had NO bug id: private comments and
    /// private attachment metadata the I5 gate removed — counted, never
    /// named.
    ///
    /// Disjoint from [`GuardInfo::suppressed_ids_count`] by
    /// construction. The two tools that feed BOTH tallies —
    /// `bug_comments` and `summarize_bug` — derive their duplicate-marker
    /// ids from the comments that SURVIVED the private filter, so a
    /// dropped comment contributes its id to nothing even when it carries
    /// one; `list_attachments`, the third id-less site, names no ids at
    /// all.
    ///
    /// No total ships beside the two (schema v2, issue #68). Addition is
    /// the reader's, and a stored total could disagree with its own
    /// parts; the split also makes the difference between the two
    /// populations legible, which the single summed field never was.
    pub suppressed_other_count: u64,
    /// The removed bug ids. Left empty by the recording call sites when
    /// [`AuditConfig::suppressed_ids`] is `false` — an empty vector then
    /// says nothing at all about how many there were, which is what
    /// [`GuardInfo::suppressed_ids_count`] is for.
    pub suppressed_ids: Vec<u64>,
    /// Names of what the response dropped (names only, never values), of
    /// two kinds: what the GUARD redacted from served bugs
    /// (`summary_view`) and the client's own presentation windows
    /// (`comments_window`, `history_window`, `attachment_window`), noted
    /// only when they actually cut something. Only the first is a guard
    /// decision, which is why the second leaves [`GuardInfo::rule`]
    /// standing — see [`AuditCell::note_redacted_client_window`].
    pub redacted_fields: Vec<String>,
    /// Search-window scan accounting, present only on calls that ran a
    /// window scan (`bugs_quicksearch`). Recorded regardless of
    /// [`AuditConfig::suppressed_ids`] — that switch elides ids, never
    /// counts. Records written before this field existed deserialize to
    /// `None`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scan: Option<ScanInfo>,
}

/// What the search-window scan behind one call examined and withheld
/// (issue #29). These numbers are exactly what I3 keeps out of the
/// response: without them a search that silently dropped many denied rows
/// is indistinguishable in the stream from one that dropped none.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ScanInfo {
    /// Upstream rows the scan examined — duplicate and id-less rows
    /// included, because each one was fetched and looked at.
    pub scanned: u64,
    /// Rows the scan dropped by verdict. Authoritative on its own, like
    /// [`GuardInfo::suppressed_ids_count`]: the dropped ids ride
    /// [`GuardInfo::suppressed_ids`] and may be elided by configuration,
    /// so consumers must never infer this count from that list.
    pub dropped: u64,
}

/// The guard's overall verdict for one tool call.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Verdict {
    /// Served in full.
    Served,
    /// Served, but with results suppressed or fields redacted.
    ServedFiltered,
    /// Denied: the client saw the uniform not-accessible response.
    Denied,
    /// Refused before any assessment (invalid arguments, disabled tool,
    /// or a fail-closed audit outage).
    Refused,
}

/// The Bugzilla side of a tool call, counted at the client's own send
/// choke point ([`bugwarden_core::client::UpstreamStats`]) — so it covers
/// the requests a handler never sees: guard classifications,
/// `resolve_caller`'s whoami, the search window's chunk scan,
/// `create_bug`'s padding classify.
///
/// Three integers by construction: nothing here can carry a URL, a
/// header, a body, an error text, or the API key (I12).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct UpstreamInfo {
    /// How many requests the client issued — REST and otherwise
    /// (`quicksearch_syntax` fetches `page.cgi`), counted whether they
    /// succeeded or not. Dispatches, not wire requests: a redirect chain
    /// reqwest follows inside one of them counts once. Never zero: a call
    /// that did not contact Bugzilla at all omits the whole block.
    pub requests: u32,
    /// HTTP status of the last request to complete. Absent where that
    /// request produced no response at all — a transport failure or a
    /// timeout — rather than reporting an earlier request's status as if
    /// it were the outcome.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub status: Option<u16>,
    /// Total time spent waiting on Bugzilla, in milliseconds: each
    /// request from hand-off to the transport until its body has been
    /// read, summed. Reading the body is time spent waiting on Bugzilla,
    /// so it counts.
    pub latency_ms: u64,
}

/// How a tool call ended, from the client's point of view.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct OutcomeInfo {
    /// Coarse outcome class.
    pub class: OutcomeClass,
    /// Total request handling time, in milliseconds.
    pub duration_ms: u64,
    /// Serialized size of the payload the server produced, in bytes: the
    /// compact JSON of the result's content array, not just the text it
    /// carries and not the enclosing result object, whose `resultType`
    /// depends on the negotiated revision (issue #145). Absent when no
    /// result exists to measure — a protocol error, or a refusal the audit
    /// gate records before it is built.
    ///
    /// Produced, not necessarily delivered: a fail-mode swap after a
    /// failed `record_event` can leave this sizing a payload the client
    /// never received — like [`OutcomeInfo::class`] and
    /// [`GuardInfo::verdict`], the record describes the call, not the
    /// wire. And the verdict does not bound it: `denied` is the *worst*
    /// verdict of the call, not a claim that nothing was served, so a
    /// multi-bug `bug_info` with one denied id still sizes the bugs it
    /// served. Reading this as "how big was the refusal" is wrong.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub response_bytes: Option<u64>,
}

/// Coarse outcome of a tool call. Note that a guard denial is `ok` here:
/// the client received a well-formed response (the uniform not-accessible
/// text); the denial itself lives in [`GuardInfo::verdict`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OutcomeClass {
    /// A well-formed result was returned.
    Ok,
    /// The call was refused (invalid arguments, disabled tool, audit
    /// outage under a closed fail mode).
    Refused,
    /// The call failed (upstream or internal error).
    Error,
}

// ---------------------------------------------------------------------------
// Errors
// ---------------------------------------------------------------------------

/// A record could not be persisted.
///
/// Deliberately content-free: an `AuditError` may end up in diagnostics,
/// so it names the failed operation and carries the underlying error, but
/// never the record that was being written.
#[derive(Debug, thiserror::Error)]
pub enum AuditError {
    /// Serializing the record to JSON failed.
    #[error("audit record serialization failed")]
    Serialize(#[source] serde_json::Error),
    /// Writing the record (or syncing it, under `fsync = true`) failed.
    #[error("audit write failed")]
    Write(#[source] std::io::Error),
    /// Rotating the audit file failed.
    #[error("audit rotation failed")]
    Rotation(#[source] std::io::Error),
    /// The export sink would not take the record — its queue is full, or
    /// it has shut down. Content-free like its siblings, and deliberately
    /// carrying no transport detail: the endpoint may not reach a
    /// diagnostic (I12).
    #[error("audit export refused the record")]
    Export,
}

impl AuditError {
    fn gap_reason(&self) -> GapReason {
        match self {
            // `Export` reports as a write error on purpose: `GapReason`
            // has no spelling for a delivery failure, and v2 declined to
            // add one. Widening a closed vocabulary is not the growth
            // rule's additive case — every reader of `audit_gap` meets
            // the new variant in a field it already reads, with no
            // `event` filter to opt out of it — so it would cost a
            // version bump for a record whose shape is unchanged. The
            // DIAGNOSTIC says which sink failed; the record says only
            // that records were lost.
            AuditError::Serialize(_) | AuditError::Write(_) | AuditError::Export => {
                GapReason::WriteError
            }
            AuditError::Rotation(_) => GapReason::RotationError,
        }
    }
}

// ---------------------------------------------------------------------------
// Sink
// ---------------------------------------------------------------------------

/// A second, LOAD-BEARING destination for records, beside (or instead of)
/// the file.
///
/// Revised 2026-08-18 (#31). Export was a best-effort copy the sink
/// ignored; it is now a sink in its own right. Every configured sink is
/// load-bearing: a record this one will not deliver puts the whole sink
/// into failure ([`AuditSink::failing`]) exactly as a failed file write
/// does, and the operator's [`FailMode`] decides what the server then does
/// about tool calls. `crate::otel` is the one implementation; the trait
/// exists so this module keeps no knowledge of OTLP.
///
/// Ordering is unchanged and still matters: when a file is configured, the
/// record reaches it FIRST and the exporter only afterwards, so the file
/// never lacks a record the collector has.
///
/// Obligations on an implementor:
///
/// - **Never block.** Every call happens with the sink's write lock held,
///   so a blocking exporter would serialize behind every audit write and
///   stall the request path. [`AuditExport::accept`] takes custody and
///   returns; delivery is reported later through
///   [`AuditExport::delivery_failing`] and [`AuditExport::take_lost`].
/// - **Never drop an audit record silently.** A record that cannot be
///   accepted must be REFUSED (so the call is gated) rather than dropped,
///   and one accepted but not delivered must be counted in
///   [`AuditExport::take_lost`] so it reaches an `audit_gap`.
/// - **`Debug` must be content-free.** [`AuditSink`] derives `Debug` and
///   would print an exporter through it, so an exporter holding
///   configuration (an endpoint, credentials) must hide it (I12).
pub trait AuditExport: Send + Sync + std::fmt::Debug {
    /// Take custody of one record, or refuse it.
    ///
    /// `event` is the record, `line` the exact bytes the file carries for
    /// it — without the terminating newline and without the leading one a
    /// torn-line repair may have prefixed. Exporting `line` verbatim is
    /// what makes the exported payload byte-equal to the file's (I12).
    ///
    /// # Errors
    ///
    /// [`ExportRefused`] when custody cannot be taken — a full queue, a
    /// shut-down pipeline. The sink turns that into [`AuditError::Export`],
    /// which the request path treats exactly like a failed write.
    fn accept(&self, event: &AuditEvent, line: &[u8]) -> Result<(), ExportRefused>;

    /// Whether delivery is currently known not to work.
    ///
    /// True from the first failed attempt until one succeeds. This is what
    /// keeps the gate closed across the whole outage rather than only at
    /// the instants a record happens to be refused.
    fn delivery_failing(&self) -> bool;

    /// Take, and reset, the count of accepted records that were never
    /// delivered. The sink folds it into its own loss accounting, so the
    /// `audit_gap` a reader sees covers both sinks.
    fn take_lost(&self) -> u64;
}

/// Why [`AuditExport::accept`] refused custody.
///
/// The reason is not distinguished on the wire: a full queue and a
/// shut-down pipeline are the same failure as far as the fail-mode gate
/// is concerned. The type exists so the refusal is not `Result<(), ()>`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ExportRefused;

#[derive(Debug)]
struct SinkState {
    /// The open live file, or `None` for a fileless (OTLP-only) sink.
    file: Option<File>,
    /// Current size of the live file, maintained across writes so
    /// rotation needs no `stat` per record.
    size: u64,
    /// Last assigned sequence number; 0 before the first record.
    seq: u64,
    /// Records lost since the last successful write; reported by the next
    /// `audit_gap` record.
    dropped: u64,
    /// Why the most recent record was lost (meaningful while
    /// `dropped > 0`).
    gap_reason: GapReason,
    /// A failed write (or sync) may have left a torn, unterminated line
    /// at the end of the live file. While set, the next written line
    /// carries a leading newline that terminates the debris —
    /// self-healing, at worst one empty line, which parsers skip (see
    /// the module docs). Cleared by the next fully successful write.
    torn: bool,
    /// When the failing-sink diagnostic was last emitted.
    last_failure_log: Option<Instant>,
}

/// Append-only JSONL writer for [`AuditEvent`] records.
///
/// All writes go through one internal mutex: `seq` assignment, rotation,
/// and the write itself are a single critical section, which is what makes
/// "file order equals `seq` order" a guarantee rather than a likelihood.
/// The API is synchronous by design (see the module docs); async callers
/// use `tokio::task::block_in_place` or `tokio::task::spawn_blocking`.
///
/// The sink assumes it is the only writer: run exactly one live sink per
/// path. Concurrent sinks on one path — in the same process or in
/// several — interleave their rename shifts and corrupt rotation.
#[derive(Debug)]
pub struct AuditSink {
    cfg: AuditConfig,
    state: Mutex<SinkState>,
    /// Load-bearing exporter attached by [`AuditSink::with_export`].
    /// See [`AuditExport`].
    export: Option<std::sync::Arc<dyn AuditExport>>,
    /// Test-only fault injection: when set, the write step fails without
    /// touching the file. Lets the gap-marker path be exercised without
    /// weakening the production API.
    #[cfg(test)]
    fail_writes: std::sync::atomic::AtomicBool,
    /// Test-only fault injection: when set, the write step writes
    /// roughly half the line and then fails — a deterministic stand-in
    /// for `ENOSPC` tearing a line mid-write.
    #[cfg(test)]
    fail_writes_partial: std::sync::atomic::AtomicBool,
    /// Test-only fault injection: when set, rotation fails before
    /// touching any file. Lets the rotation-failure gap path be
    /// exercised.
    #[cfg(test)]
    fail_rotation: std::sync::atomic::AtomicBool,
}

impl AuditSink {
    /// Open (or create) the audit file and return a ready sink.
    ///
    /// - Refuses if `cfg.path` exists and is a symlink. Best-effort by
    ///   design: the audit directory is expected to be operator-owned, so
    ///   the pre-check catches misconfiguration (a link left behind, a
    ///   path pointed somewhere surprising) rather than defending against
    ///   a racing local attacker.
    /// - Creates the parent directory (mode 0700) if missing.
    /// - Opens the file append-only (`O_WRONLY|O_APPEND|O_CREAT`), mode
    ///   0600 when created.
    ///
    /// Errors carry the offending path; they never carry record content
    /// (there is none yet).
    pub fn open(cfg: AuditConfig) -> anyhow::Result<AuditSink> {
        // A hand-constructed config could bypass load-time validation, so
        // re-validate here; the check is free.
        cfg.validate()?;
        // A fileless sink touches no filesystem at all: no symlink probe,
        // no directory creation, no open. That is what lets a container
        // run OTLP-only with nothing writable mounted.
        let (file, size) = match &cfg.path {
            None => (None, 0),
            Some(path) => {
                if let Ok(meta) = std::fs::symlink_metadata(path) {
                    if meta.file_type().is_symlink() {
                        anyhow::bail!(
                            "audit file {} is a symlink; refusing to follow it — point \
                             `path` at a regular file in an operator-owned directory",
                            path.display()
                        );
                    }
                }
                if let Some(parent) = path.parent() {
                    if !parent.as_os_str().is_empty() {
                        let mut builder = std::fs::DirBuilder::new();
                        builder.recursive(true);
                        #[cfg(unix)]
                        {
                            use std::os::unix::fs::DirBuilderExt as _;
                            builder.mode(0o700);
                        }
                        builder.create(parent).with_context(|| {
                            format!("failed to create audit directory {}", parent.display())
                        })?;
                    }
                }
                let file = open_live_file(path)
                    .with_context(|| format!("failed to open audit file {}", path.display()))?;
                let size = file
                    .metadata()
                    .with_context(|| format!("failed to stat audit file {}", path.display()))?
                    .len();
                (Some(file), size)
            }
        };
        Ok(AuditSink {
            cfg,
            export: None,
            state: Mutex::new(SinkState {
                file,
                size,
                seq: 0,
                dropped: 0,
                gap_reason: GapReason::WriteError,
                torn: false,
                last_failure_log: None,
            }),
            #[cfg(test)]
            fail_writes: std::sync::atomic::AtomicBool::new(false),
            #[cfg(test)]
            fail_writes_partial: std::sync::atomic::AtomicBool::new(false),
            #[cfg(test)]
            fail_rotation: std::sync::atomic::AtomicBool::new(false),
        })
    }

    /// Attach the load-bearing exporter that receives every record this
    /// sink persists; see [`AuditExport`]. A fileless sink MUST have one —
    /// without it records would vanish silently — and `main`'s sink
    /// selection is what guarantees that pairing.
    ///
    /// Consuming, so the exporter can only be attached while the sink is
    /// still being built at startup — a running sink's export side never
    /// changes, and the request path cannot reach it.
    #[must_use]
    pub fn with_export(mut self, export: std::sync::Arc<dyn AuditExport>) -> AuditSink {
        self.export = Some(export);
        self
    }

    /// Persist one record; returns its assigned `seq`.
    ///
    /// Blocks until the record is written (and synced, under `fsync`). If
    /// earlier records were lost, an `audit_gap` record reporting them is
    /// written first — and must itself succeed — before the loss counter
    /// resets; the gap record is stamped with this call's `session`.
    ///
    /// On failure the record is counted as dropped, a rate-limited
    /// `tracing::error!` diagnostic is emitted (first failure immediately,
    /// then at most one per minute), and the error is
    /// returned so the caller can apply the configured [`FailMode`]. A
    /// failed record consumes no `seq` — with one deliberate exception:
    /// under [`AuditConfig::fsync`], a `sync_data` failure after a
    /// successful write counts the record as dropped even though its
    /// bytes (seq included) may already be in the file, so that seq can
    /// appear twice on disk and the following `audit_gap` may
    /// over-report the loss. Over-reporting, never under-reporting, is
    /// the chosen direction.
    pub fn record(&self, kind: AuditEventKind, session: SessionInfo) -> Result<u64, AuditError> {
        debug_assert!(
            self.cfg.path.is_some() || self.export.is_some(),
            "a fileless audit sink must have an exporter; records would vanish"
        );
        // A poisoned mutex means a writer panicked; the size/seq state is
        // still usable (worst case an early rotation), so keep going —
        // dropping audit coverage over a bookkeeping wobble would be the
        // worse trade.
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // Records the exporter accepted and then failed to deliver are
        // losses of this sink like any other, so they join the same
        // counter and surface through the same `audit_gap`. Folded before
        // the gap check below, so an export outage is reported in the very
        // next record rather than one later.
        self.absorb_export_losses(&mut state);
        if state.dropped > 0 {
            let gap = AuditEventKind::AuditGap(AuditGapEvent {
                dropped: state.dropped,
                reason: state.gap_reason,
            });
            match self.write_event(&mut state, gap, session.clone()) {
                Ok(_) => state.dropped = 0,
                Err(err) => {
                    // The gap record did not make it out either. The
                    // caller's record was never attempted, so it is lost
                    // with the rest and joins the count.
                    state.dropped = state.dropped.saturating_add(1);
                    state.gap_reason = err.gap_reason();
                    self.log_failure(&mut state, &err);
                    return Err(err);
                }
            }
        }
        match self.write_event(&mut state, kind, session) {
            Ok(seq) => Ok(seq),
            Err(err) => {
                state.dropped = state.dropped.saturating_add(1);
                state.gap_reason = err.gap_reason();
                self.log_failure(&mut state, &err);
                Err(err)
            }
        }
    }

    /// Stamp, serialize, rotate if due, write, sync, account. Called with
    /// the state lock held.
    fn write_event(
        &self,
        state: &mut SinkState,
        kind: AuditEventKind,
        session: SessionInfo,
    ) -> Result<u64, AuditError> {
        let seq = state.seq + 1;
        let event = AuditEvent {
            v: SCHEMA_VERSION,
            ts: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
            seq,
            session,
            kind,
        };
        let mut line = Vec::new();
        // A failed write may have left a torn, unterminated line at the
        // end of the file; a leading newline terminates it so the stream
        // self-heals (worst case one empty line, which parsers skip —
        // see the module docs).
        let healed = state.torn;
        if healed {
            line.push(b'\n');
        }
        serde_json::to_writer(&mut line, &event).map_err(AuditError::Serialize)?;
        line.push(b'\n');
        let len = line.len() as u64;
        // Rotate BEFORE writing, so the limit is never exceeded. A single
        // record larger than the limit is written into an empty live file
        // as-is: rotating an empty file would discard nothing and loop.
        if self.cfg.rotate_max_bytes > 0
            && state.size > 0
            && state.size + len > self.cfg.rotate_max_bytes
        {
            self.rotate(state).map_err(AuditError::Rotation)?;
        }
        #[cfg(test)]
        if self.fail_writes.load(std::sync::atomic::Ordering::Relaxed) {
            return Err(AuditError::Write(std::io::Error::other(
                "injected write failure (test)",
            )));
        }
        #[cfg(test)]
        if self
            .fail_writes_partial
            .load(std::sync::atomic::Ordering::Relaxed)
        {
            // Write roughly half the line, then fail: a deterministic
            // stand-in for ENOSPC tearing a line mid-write.
            if let Some(file) = state.file.as_mut() {
                file.write_all(&line[..line.len() / 2])
                    .map_err(AuditError::Write)?;
            }
            state.torn = true;
            return Err(AuditError::Write(std::io::Error::other(
                "injected partial write failure (test)",
            )));
        }
        if let Some(file) = state.file.as_mut() {
            file.write_all(&line).map_err(|e| {
                // An unknown number of bytes made it out: assume a torn
                // line.
                state.torn = true;
                AuditError::Write(e)
            })?;
            if self.cfg.fsync {
                file.sync_data().map_err(|e| {
                    // The line is complete in the page cache but may not
                    // have reached the disk; treat it as torn so the next
                    // record re-terminates whatever survives.
                    state.torn = true;
                    AuditError::Write(e)
                })?;
            }
            state.size += len;
        }
        state.torn = false;
        // The file has the record and its `seq` is spent, whatever the
        // export does next: a `seq` reused after a rejected export would
        // put two different records on one number in the file.
        state.seq = seq;
        if let Some(export) = &self.export {
            // AFTER the write, never before: the file, when there is one,
            // is authoritative, and a record it never took must not exist
            // anywhere else. The slice strips the repair newline this
            // record may have been prefixed with and the one terminating
            // it, so what the exporter gets is byte-for-byte the record
            // line the file holds. Still under the state lock, so export
            // order equals `seq` order — accepting only queues.
            let start = usize::from(healed);
            // Load-bearing since 2026-08-18 (#31): a record the exporter
            // cannot even accept is a record that will not be delivered,
            // and saying so here is what puts the call in front of the
            // same fail-mode gate a failed file write does. The file, if
            // any, keeps the record — the loss accounting over-reports
            // rather than under-reports, as it does under `fsync`.
            export
                .accept(&event, &line[start..line.len() - 1])
                .map_err(|ExportRefused| AuditError::Export)?;
        }
        Ok(seq)
    }

    /// Numbered shift rotation: `<path>.N` → `<path>.N+1` from the
    /// highest kept slot down, live file → `<path>.1`, live file
    /// recreated mode 0600, parent directory fsynced so the renames are
    /// durable. Called with the state lock held.
    fn rotate(&self, state: &mut SinkState) -> std::io::Result<()> {
        #[cfg(test)]
        if self
            .fail_rotation
            .load(std::sync::atomic::Ordering::Relaxed)
        {
            return Err(std::io::Error::other("injected rotation failure (test)"));
        }
        // Unreachable for a fileless sink: `write_event` only rotates when
        // it holds a file, and `size` never leaves 0 without one.
        let Some(live) = &self.cfg.path else {
            return Ok(());
        };
        // The oldest kept slot would shift past the cap: delete it.
        match std::fs::remove_file(rotated_path(live, self.cfg.rotate_keep)) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e),
        }
        for n in (2..=self.cfg.rotate_keep).rev() {
            match std::fs::rename(rotated_path(live, n - 1), rotated_path(live, n)) {
                Ok(()) => {}
                // A slot that does not exist yet has nothing to shift.
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(e),
            }
        }
        match std::fs::rename(live, rotated_path(live, 1)) {
            Ok(()) => {}
            // Recovery from a previous partial rotation (renamed but not
            // recreated): nothing to move, just recreate below.
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e),
        }
        state.file = Some(open_live_file(live)?);
        state.size = 0;
        #[cfg(unix)]
        if let Some(parent) = live.parent() {
            let dir = if parent.as_os_str().is_empty() {
                Path::new(".")
            } else {
                parent
            };
            File::open(dir)?.sync_all()?;
        }
        Ok(())
    }

    /// Rate-limited `tracing::error!` about a failing sink. Called with
    /// the state lock held. Carries the error and the drop count — never
    /// record content.
    fn log_failure(&self, state: &mut SinkState, err: &AuditError) {
        let due = state
            .last_failure_log
            .is_none_or(|last| last.elapsed() >= FAILURE_LOG_INTERVAL);
        if due {
            state.last_failure_log = Some(Instant::now());
            tracing::error!(
                error = ?err,
                dropped = state.dropped,
                path = self.cfg.path.as_ref().map(|p| p.display().to_string()),
                "audit record could not be persisted; records are being dropped"
            );
        }
    }

    /// Whether the sink is currently in failure: at least one record has
    /// been lost since the last success on ANY configured destination, or
    /// export delivery is known not to be working. The request path
    /// consults this before dispatching a tool call, so a sink that is
    /// already down can hold back further unaudited work under the closed
    /// fail modes instead of discovering the outage one record at a time.
    ///
    /// The second condition is what makes an export outage as visible as a
    /// full disk. Loss counters clear as soon as their `audit_gap` is
    /// written, and for the exporter "written" means "accepted for
    /// delivery" — so a sink judged only by its counters would report
    /// healthy the instant it queued the gap record, and flap once per
    /// batch for the whole outage. Delivery health does not clear until a
    /// request actually succeeds.
    pub fn failing(&self) -> bool {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // Also here, not only in `record`: the pre-dispatch gate asks this
        // question far more often than it writes, and an outage that only
        // became visible between two records must close the gate at the
        // first ask.
        self.absorb_export_losses(&mut state);
        state.dropped > 0
            || self
                .export
                .as_ref()
                .is_some_and(|export| export.delivery_failing())
    }

    /// Move any undelivered-record count out of the exporter and into this
    /// sink's own loss accounting. Called with the state lock held.
    fn absorb_export_losses(&self, state: &mut SinkState) {
        let Some(export) = &self.export else {
            return;
        };
        let lost = export.take_lost();
        if lost > 0 {
            state.dropped = state.dropped.saturating_add(lost);
            state.gap_reason = GapReason::WriteError;
        }
    }

    /// Whether records may carry suppressed bug ids
    /// ([`AuditConfig::suppressed_ids`]); the recording call site enforces
    /// the flag when building [`GuardInfo`].
    pub fn suppressed_ids(&self) -> bool {
        self.cfg.suppressed_ids
    }

    /// Test-only fault injection; see the `fail_writes` field. Crate
    /// visibility: the request-path fail-mode tests in `server.rs` drive
    /// a real sink into failure through this same hook.
    #[cfg(test)]
    pub(crate) fn set_fail_writes(&self, fail: bool) {
        self.fail_writes
            .store(fail, std::sync::atomic::Ordering::Relaxed);
    }

    /// Test-only fault injection; see the `fail_writes_partial` field.
    #[cfg(test)]
    fn set_fail_writes_partial(&self, fail: bool) {
        self.fail_writes_partial
            .store(fail, std::sync::atomic::Ordering::Relaxed);
    }

    /// Test-only fault injection; see the `fail_rotation` field.
    #[cfg(test)]
    fn set_fail_rotation(&self, fail: bool) {
        self.fail_rotation
            .store(fail, std::sync::atomic::Ordering::Relaxed);
    }
}

/// Open the live audit file append-only, creating it mode 0600 if absent.
fn open_live_file(path: &Path) -> std::io::Result<File> {
    let mut opts = OpenOptions::new();
    opts.append(true).create(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt as _;
        opts.mode(0o600);
    }
    opts.open(path)
}

/// `<path>.<n>`, the n-th rotated file.
fn rotated_path(live: &Path, n: u32) -> PathBuf {
    let mut os = live.as_os_str().to_os_string();
    os.push(format!(".{n}"));
    PathBuf::from(os)
}

// ---------------------------------------------------------------------------
// Request-path wiring
// ---------------------------------------------------------------------------

impl FailMode {
    /// The fail mode in force: the operator's explicit choice when the
    /// configuration sets one, else the transport-derived default.
    pub fn resolve(explicit: Option<FailMode>, transport: TransportKind) -> FailMode {
        explicit.unwrap_or_else(|| FailMode::default_for(transport))
    }

    /// The default fail mode for a transport, used when the audit
    /// configuration leaves `fail_mode` unset: `stdio` (a local,
    /// single-operator session) defaults to `open` — availability for a
    /// single user; `http` (remote clients) defaults to `closed_all` —
    /// accountability for a fleet.
    pub fn default_for(transport: TransportKind) -> FailMode {
        match transport {
            TransportKind::Stdio => FailMode::Open,
            TransportKind::Http => FailMode::ClosedAll,
        }
    }
}

/// Everything the request path needs to write audit records: the open
/// sink, the resolved fail mode, and the startup-computed identifiers
/// records carry. Built once at startup, shared behind an `Arc`.
#[derive(Debug)]
pub struct AuditState {
    /// The open sink; every record goes through it.
    pub sink: AuditSink,
    /// What happens to tool calls while records cannot be persisted.
    /// Already resolved ([`FailMode::resolve`]): the explicit configured
    /// value, else the transport default.
    pub fail_mode: FailMode,
    /// `sha256:<hex>` digest of the policy file in force, stamped into
    /// [`GuardInfo::policy_hash`]. `None` when the built-in default
    /// policy applies (there is no file to hash).
    pub policy_hash: Option<String>,
    /// Session id stamped into stdio records. One process is one stdio
    /// session, so `<pid>-<unix epoch secs at startup>` groups those
    /// records the way the transport-minted id groups http ones (see
    /// [`crate::http_session`] — never the request header, #180).
    ///
    /// Stamped UNCONDITIONALLY, including on the 2026-07-28 per-request
    /// lifecycle, where the http path records no id at all: there the only
    /// candidate is a forgeable request header, here the value is minted
    /// from the process and cannot be forged. The consequence is that a
    /// stdio client which never sends `initialize` produces a run of
    /// records sharing an id that no `initialize` anchors — see
    /// [`SessionInfo::id`], which is where that limit is stated.
    pub stdio_session_id: String,
}

/// The `sha256:<hex>` digest of a policy document's bytes — the value
/// [`AuditState::policy_hash`] holds and [`GuardInfo::policy_hash`]
/// carries. The digest is over the raw file bytes, not any parsed form,
/// so a record ties to the exact document the operator deployed.
pub fn policy_hash_of(bytes: &[u8]) -> String {
    use std::fmt::Write as _;

    // Hex-encode the bytes here rather than formatting the digest with
    // `{:x}`: the `LowerHex` impl belongs to generic-array's
    // `GenericArray`, which is sha2 0.10's output type, and 0.11's
    // hybrid-array `Array` has none — so formatting the digest itself
    // pins the crate to a major version.
    let digest = Sha256::digest(bytes);
    let mut hash = String::with_capacity("sha256:".len() + digest.len() * 2);
    hash.push_str("sha256:");
    for byte in &digest {
        // Writing to a String cannot fail.
        let _ = write!(hash, "{byte:02x}");
    }
    hash
}

impl AuditState {
    /// Assemble the request-path state around an open sink.
    pub fn new(sink: AuditSink, fail_mode: FailMode, policy_hash: Option<String>) -> AuditState {
        let epoch_secs = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        AuditState {
            sink,
            fail_mode,
            policy_hash,
            stdio_session_id: format!("{}-{epoch_secs}", std::process::id()),
        }
    }
}

/// Severity of a verdict for the worst-wins merge:
/// `Denied > Refused > ServedFiltered > Served`.
fn verdict_rank(verdict: Verdict) -> u8 {
    match verdict {
        Verdict::Served => 0,
        Verdict::ServedFiltered => 1,
        Verdict::Refused => 2,
        Verdict::Denied => 3,
    }
}

#[derive(Debug, Default)]
struct CellState {
    verdict: Option<Verdict>,
    rule: Option<String>,
    suppressed_ids: std::collections::BTreeSet<u64>,
    /// Suppressions that have no bug id (private comments or attachment
    /// metadata filtered out): counted, never named.
    suppressed_extra: u64,
    redacted_fields: std::collections::BTreeSet<String>,
    /// Search-window scan accounting ([`AuditCell::note_scan`]).
    scan: Option<ScanInfo>,
    upstream: Option<UpstreamInfo>,
}

/// Per-request enrichment cell for one tool call's audit record.
///
/// The wrapper that owns the record inserts one cell into the request's
/// extensions before dispatch; the guard call sites inside the tool note
/// what was decided; the wrapper drains the cell into [`GuardInfo`] after
/// the tool returns. A tool that never touches the cell (it performs no
/// guard assessment) yields no `GuardInfo` at all. Multi-id calls note
/// once per id and the cell merges: ONE record per call, worst verdict,
/// with suppression carrying the per-id detail.
///
/// Interior mutability via a `Mutex` with the same poison posture as the
/// sink: a panicked noter loses nothing worth dropping enrichment over.
#[derive(Debug, Default)]
pub struct AuditCell {
    state: Mutex<CellState>,
}

impl AuditCell {
    fn lock(&self) -> std::sync::MutexGuard<'_, CellState> {
        self.state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// Worst-wins verdict merge. The rule follows the deciding verdict: a
    /// strict upgrade replaces the stored rule with the new one (present
    /// or not — a rule-less verdict clears it), an equally severe
    /// verdict may only fill an empty slot (first rule recorded wins),
    /// and a less severe verdict changes nothing.
    fn merge(&self, verdict: Verdict, rule: Option<&str>) {
        self.merge_inner(verdict, rule, false);
    }

    /// [`AuditCell::merge`], with `keep_rule` suppressing every write to
    /// the stored rule. One body, so the client-window carve-out cannot
    /// drift from the merge it is a carve-out of.
    fn merge_inner(&self, verdict: Verdict, rule: Option<&str>, keep_rule: bool) {
        let mut state = self.lock();
        let rank = verdict_rank(verdict);
        match state.verdict.map(verdict_rank) {
            Some(current) if rank < current => {}
            Some(current) if rank == current => {
                if state.rule.is_none() && !keep_rule {
                    state.rule = rule.map(str::to_owned);
                }
            }
            // Absent, or a strict upgrade: this verdict decides the call.
            _ => {
                state.verdict = Some(verdict);
                if !keep_rule {
                    state.rule = rule.map(str::to_owned);
                }
            }
        }
    }

    /// Note a verdict no single rule decided: a refusal that never
    /// consulted one, a search or create gate judging the request as a
    /// whole, an id the guard produced no matching decision for, or a
    /// withheld attachment — [`GuardInfo::rule`] carries the full list.
    /// NOT for a policy-default decision: the default decides under the
    /// rule name `"default"` and goes through
    /// [`AuditCell::note_verdict_rule`] like any other.
    pub fn note_verdict(&self, verdict: Verdict) {
        self.merge(verdict, None);
    }

    /// Note a verdict together with the rule that decided it.
    pub fn note_verdict_rule(&self, verdict: Verdict, rule: &str) {
        self.merge(verdict, Some(rule));
    }

    /// Note bug ids the guard hid in this call (unions across notes).
    /// Suppression implies at least a filtered serve, so the verdict is
    /// upgraded to `ServedFiltered` through the worst-wins merge — a
    /// record can never claim a clean `Served` while carrying
    /// suppressions.
    pub fn note_suppressed(&self, ids: impl IntoIterator<Item = u64>) {
        self.merge(Verdict::ServedFiltered, None);
        self.lock().suppressed_ids.extend(ids);
    }

    /// Note `n` suppressions that carry no bug id (private comments or
    /// attachment metadata filtered out). Adds across notes; upgrades the
    /// verdict exactly as [`AuditCell::note_suppressed`] does.
    ///
    /// This tally and [`AuditCell::note_suppressed`]'s id set must stay
    /// disjoint populations: they ship as two fields a reader ADDS
    /// ([`GuardInfo::suppressed_other_count`] and
    /// [`GuardInfo::suppressed_ids_count`]), so a call site that counted
    /// an item here and also named it there would count it twice.
    pub fn note_suppressed_count(&self, n: u64) {
        if n == 0 {
            return;
        }
        self.merge(Verdict::ServedFiltered, None);
        let mut state = self.lock();
        state.suppressed_extra = state.suppressed_extra.saturating_add(n);
    }

    /// Note the search-window scan's accounting (issue #29): how many
    /// upstream rows it examined and how many it dropped by verdict. One
    /// scan per call by construction; if noted twice, the last note wins.
    /// A scan that dropped rows IS a filtered serve, so `dropped > 0`
    /// upgrades the verdict exactly as [`AuditCell::note_suppressed`]
    /// does; a clean scan records its numbers without touching the
    /// verdict at all. The dropped IDS are not this method's business —
    /// the call site feeds them to [`AuditCell::note_suppressed`], where
    /// the existing `suppressed_ids` config switch governs them.
    pub fn note_scan(&self, scanned: u64, dropped: u64) {
        if dropped > 0 {
            self.merge(Verdict::ServedFiltered, None);
        }
        self.lock().scan = Some(ScanInfo { scanned, dropped });
    }

    /// Note a field (or view marker) the GUARD redacted from served bugs
    /// — a decision no single rule made, so the merge carries no rule of
    /// its own. What that does to an already-recorded rule follows
    /// `AuditCell::merge` and nothing else: it CLEARS when this note is
    /// what upgrades the verdict (`list_attachments` dropping private
    /// metadata off a plain serve), and leaves it standing at equal rank
    /// (`bug_info` records the `ServedFiltered` verdict AND its rule
    /// first, then names `summary_view` here). A client-requested
    /// presentation window is not a guard redaction at all; it goes
    /// through [`AuditCell::note_redacted_client_window`].
    pub fn note_redacted(&self, field: &str) {
        self.merge(Verdict::ServedFiltered, None);
        self.lock().redacted_fields.insert(field.to_owned());
    }

    /// Note a CLIENT-requested presentation window that dropped content
    /// (`head`/`tail`/length caps). Records the field and upgrades to
    /// `ServedFiltered` exactly as [`AuditCell::note_redacted`] does, but
    /// never writes the rule at all — not even on the strict upgrade
    /// where `AuditCell::merge` would clear it. The client asked for
    /// less; no rule decided anything, and clearing would report a call a
    /// rule DID grant as rule-less (issue #67). Order-independent by
    /// construction: it cannot clear a rule a later guard note records,
    /// and a guard note landing after it merges by the ordinary rules.
    pub fn note_redacted_client_window(&self, field: &str) {
        self.merge_inner(Verdict::ServedFiltered, None, true);
        self.lock().redacted_fields.insert(field.to_owned());
    }

    /// Note the Bugzilla side of the call. Called once per tool call by
    /// the `call_tool` wrapper, from the accounting the core client
    /// collected over the dispatch — never by a tool.
    pub fn note_upstream(&self, upstream: UpstreamInfo) {
        self.lock().upstream = Some(upstream);
    }

    /// Take the noted upstream info, if any.
    pub fn take_upstream(&self) -> Option<UpstreamInfo> {
        self.lock().upstream.take()
    }

    /// Drain the cell into the record's [`GuardInfo`]. `None` when no
    /// verdict was ever noted — the tool performed no guard assessment.
    /// When `suppressed_ids_cfg` is `false` the ids are dropped and only
    /// the counts ship ([`AuditConfig::suppressed_ids`]). The two tallies
    /// are projected into their own fields (issue #68, schema v2), never
    /// summed here: they are disjoint populations, and the reader adds
    /// them if it wants a total.
    pub fn into_guard_info(
        &self,
        policy_hash: Option<&str>,
        suppressed_ids_cfg: bool,
    ) -> Option<GuardInfo> {
        let state = std::mem::take(&mut *self.lock());
        let verdict = state.verdict?;
        // From the SET, before the config elision below empties the
        // emitted vector — deriving it from that vector would zero the
        // count exactly where it is the only signal left.
        let suppressed_ids_count = state.suppressed_ids.len() as u64;
        Some(GuardInfo {
            verdict,
            rule: state.rule,
            policy_hash: policy_hash.map(str::to_owned),
            suppressed_ids_count,
            suppressed_other_count: state.suppressed_extra,
            suppressed_ids: if suppressed_ids_cfg {
                state.suppressed_ids.into_iter().collect()
            } else {
                Vec::new()
            },
            redacted_fields: state.redacted_fields.into_iter().collect(),
            // Counts, not ids: the suppressed_ids switch never touches it.
            scan: state.scan,
        })
    }
}

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

    // ---------- fixtures ----------

    const TS: &str = "2026-02-03T04:05:06.789Z";

    fn session_stdio() -> SessionInfo {
        SessionInfo {
            id: None,
            transport: TransportKind::Stdio,
            remote: None,
        }
    }

    fn session_http() -> SessionInfo {
        SessionInfo {
            id: Some("sess-1".into()),
            transport: TransportKind::Http,
            remote: Some("192.0.2.7:52611".into()),
        }
    }

    /// A `tool_call` with every schema field populated — the two
    /// reserved identity slots included, which no record this build
    /// writes carries. An example value does not bind their contract;
    /// filling them here is what makes the goldens a shape test.
    fn sample_tool_call() -> AuditEventKind {
        AuditEventKind::ToolCall(Box::new(ToolCallEvent {
            client: ClientInfo {
                name: Some("example-agent".into()),
                version: Some("1.4.2".into()),
                principal: Some("alice@example.org".into()),
                work_context: Some("SUSE:Maintenance:12345:678901".into()),
            },
            trace: Some(TraceContext {
                trace_id: "0af7651916cd43dd8448eb211c80319c".into(),
                span_id: "b7ad6b7169203331".into(),
            }),
            request: RequestInfo {
                tool: "bug_info".into(),
                id: Some("3".into()),
                params: BTreeMap::from([
                    ("id".to_string(), json!(1290042)),
                    ("include_private".to_string(), json!(false)),
                ]),
            },
            guard: Some(GuardInfo {
                verdict: Verdict::ServedFiltered,
                rule: Some("group-restricted".into()),
                policy_hash: Some("2b6e8f5d1c9a4e70".into()),
                suppressed_ids_count: 2,
                // Deliberately NOT equal to the ids count: a swap of the
                // two fields must change the golden's bytes.
                suppressed_other_count: 3,
                suppressed_ids: vec![1290040, 1290041],
                redacted_fields: vec!["cc".into(), "flags".into()],
                scan: Some(ScanInfo {
                    scanned: 200,
                    dropped: 2,
                }),
            }),
            upstream: Some(UpstreamInfo {
                requests: 1,
                status: Some(200),
                latency_ms: 48,
            }),
            outcome: OutcomeInfo {
                class: OutcomeClass::Ok,
                duration_ms: 52,
                response_bytes: Some(1462),
            },
        }))
    }

    fn sample_initialize() -> AuditEventKind {
        AuditEventKind::Initialize(InitializeEvent {
            client: ClientInfo {
                name: Some("example-agent".into()),
                version: Some("1.4.2".into()),
                principal: None,
                work_context: None,
            },
            protocol_version: Some("2025-06-18".into()),
        })
    }

    fn sample_gap() -> AuditEventKind {
        AuditEventKind::AuditGap(AuditGapEvent {
            dropped: 3,
            reason: GapReason::WriteError,
        })
    }

    fn envelope(seq: u64, session: SessionInfo, kind: AuditEventKind) -> AuditEvent {
        AuditEvent {
            v: SCHEMA_VERSION,
            ts: TS.to_string(),
            seq,
            session,
            kind,
        }
    }

    fn test_cfg(path: PathBuf) -> AuditConfig {
        AuditConfig {
            path: Some(path),
            fsync: false,
            fail_mode: None,
            rotate_max_bytes: 0,
            rotate_keep: 8,
            suppressed_ids: true,
        }
    }

    /// The live file of a config built by [`test_cfg`], which always
    /// names one.
    fn live_path(cfg: &AuditConfig) -> &Path {
        cfg.path.as_deref().expect("test config names a file")
    }

    /// Parse every line of an audit file. Empty lines are skipped, as
    /// the module docs require of parsers: a recovered write outage may
    /// leave one behind.
    fn read_events(path: &Path) -> Vec<AuditEvent> {
        let s = std::fs::read_to_string(path).expect("audit file must be readable");
        s.lines()
            .filter(|l| !l.is_empty())
            .map(|l| serde_json::from_str(l).expect("every audit line must parse as AuditEvent"))
            .collect()
    }

    /// Assert the live file and every rotated file respect
    /// `rotate_max_bytes`. Rotation happens before the write, so the
    /// limit is a hard bound (a single oversized record is the one
    /// documented exception, tested separately).
    fn assert_files_within_rotate_limit(cfg: &AuditConfig) {
        let mut paths = vec![live_path(cfg).to_path_buf()];
        paths.extend((1..=cfg.rotate_keep).map(|n| rotated_path(live_path(cfg), n)));
        for p in paths {
            if let Ok(meta) = std::fs::metadata(&p) {
                assert!(
                    meta.len() <= cfg.rotate_max_bytes,
                    "{} is {} bytes, over rotate_max_bytes = {}",
                    p.display(),
                    meta.len(),
                    cfg.rotate_max_bytes
                );
            }
        }
    }

    // ---------- schema goldens (v2 stability gate) ----------
    //
    // These strings ARE the wire format. A failure here means the schema
    // changed; that requires a deliberate decision, not an updated
    // constant. Never paste serializer output into one of these — that
    // can only enshrine whatever the code does, bug included; derive the
    // new bytes by hand from the old ones and let the test judge the
    // code.
    //
    // WITHIN v2 one shape of change is allowed: adding a field that is
    // optional and absent when unset, or adding a record kind (see
    // `SCHEMA_VERSION`). An added FIELD updates the golden AND must pin
    // the prior byte sequence as a GOLDEN_*_PRE_* constant with a test
    // proving old lines still deserialize; an added KIND has no prior
    // bytes to pin and simply gains a golden of its own. ACROSS a bump the discipline
    // inverts: the prior version's line stays pinned — here
    // `GOLDEN_TOOL_CALL_V1` — with a test proving the current reader
    // REJECTS it. Removing, renaming, reordering or retyping a field is
    // what forces that bump.

    const GOLDEN_TOOL_CALL: &str = concat!(
        r#"{"v":2,"ts":"2026-02-03T04:05:06.789Z","seq":7,"#,
        r#""session":{"id":"sess-1","transport":"http","remote":"192.0.2.7:52611"},"#,
        r#""event":"tool_call","#,
        r#""client":{"name":"example-agent","version":"1.4.2","principal":"alice@example.org","work_context":"SUSE:Maintenance:12345:678901"},"#,
        r#""trace":{"trace_id":"0af7651916cd43dd8448eb211c80319c","span_id":"b7ad6b7169203331"},"#,
        r#""request":{"tool":"bug_info","id":"3","params":{"id":1290042,"include_private":false}},"#,
        r#""guard":{"verdict":"served_filtered","rule":"group-restricted","policy_hash":"2b6e8f5d1c9a4e70","#,
        r#""suppressed_ids_count":2,"suppressed_other_count":3,"suppressed_ids":[1290040,1290041],"#,
        r#""redacted_fields":["cc","flags"],"scan":{"scanned":200,"dropped":2}},"#,
        r#""upstream":{"requests":1,"status":200,"latency_ms":48},"#,
        r#""outcome":{"class":"ok","duration_ms":52,"response_bytes":1462}}"#,
    );

    /// The last `tool_call` byte sequence schema v1 produced, pinned so
    /// the break at v2 is tested rather than assumed. A v2 reader must
    /// REJECT it: `deny_unknown_fields` refuses `guard.suppressed_count`,
    /// and there is no honest way to accept it — the field summed two
    /// populations that cannot be separated after the fact.
    const GOLDEN_TOOL_CALL_V1: &str = concat!(
        r#"{"v":1,"ts":"2026-02-03T04:05:06.789Z","seq":7,"#,
        r#""session":{"id":"sess-1","transport":"http","remote":"192.0.2.7:52611"},"#,
        r#""event":"tool_call","#,
        r#""client":{"name":"example-agent","version":"1.4.2","principal":"alice@example.org"},"#,
        r#""trace":{"trace_id":"0af7651916cd43dd8448eb211c80319c","span_id":"b7ad6b7169203331"},"#,
        r#""request":{"tool":"bug_info","id":"3","params":{"id":1290042,"include_private":false}},"#,
        r#""guard":{"verdict":"served_filtered","rule":"group-restricted","policy_hash":"2b6e8f5d1c9a4e70","#,
        r#""suppressed_count":2,"suppressed_ids":[1290040,1290041],"redacted_fields":["cc","flags"],"#,
        r#""scan":{"scanned":200,"dropped":2}},"#,
        r#""upstream":{"requests":1,"status":200,"latency_ms":48},"#,
        r#""outcome":{"class":"ok","duration_ms":52,"response_bytes":1462}}"#,
    );

    /// The v1 lines OUTSIDE the break: a `tool_call` from a tool that
    /// consulted no guard (`bug_url`), and v1's `initialize` and
    /// `audit_gap`, whose payloads v2 did not touch. None of them names a
    /// field v2 removed, so all three still parse and still read `v: 1`.
    /// That is the exact scope of the break, pinned here so widening it
    /// takes a deliberate edit.
    const V1_LINES_OUTSIDE_THE_BREAK: [&str; 3] = [
        concat!(
            r#"{"v":1,"ts":"2026-02-03T04:05:06.789Z","seq":7,"#,
            r#""session":{"id":"sess-1","transport":"http","remote":"192.0.2.7:52611"},"#,
            r#""event":"tool_call","#,
            r#""client":{"name":"example-agent","version":"1.4.2"},"#,
            r#""request":{"tool":"bug_url","id":"3","params":{"id":1290042}},"#,
            r#""outcome":{"class":"ok","duration_ms":1}}"#,
        ),
        concat!(
            r#"{"v":1,"ts":"2026-02-03T04:05:06.789Z","seq":1,"session":{"transport":"stdio"},"#,
            r#""event":"initialize","client":{"name":"example-agent","version":"1.4.2"},"#,
            r#""protocol_version":"2025-06-18"}"#,
        ),
        concat!(
            r#"{"v":1,"ts":"2026-02-03T04:05:06.789Z","seq":9,"session":{"transport":"stdio"},"#,
            r#""event":"audit_gap","dropped":3,"reason":"write_error"}"#,
        ),
    ];

    const GOLDEN_INITIALIZE: &str = concat!(
        r#"{"v":2,"ts":"2026-02-03T04:05:06.789Z","seq":1,"session":{"transport":"stdio"},"#,
        r#""event":"initialize","client":{"name":"example-agent","version":"1.4.2"},"#,
        r#""protocol_version":"2025-06-18"}"#,
    );

    const GOLDEN_AUDIT_GAP: &str = concat!(
        r#"{"v":2,"ts":"2026-02-03T04:05:06.789Z","seq":9,"session":{"transport":"stdio"},"#,
        r#""event":"audit_gap","dropped":3,"reason":"write_error"}"#,
    );

    #[test]
    fn golden_tool_call_json() {
        let event = envelope(7, session_http(), sample_tool_call());
        assert_eq!(serde_json::to_string(&event).unwrap(), GOLDEN_TOOL_CALL);
    }

    #[test]
    fn golden_initialize_json() {
        let event = envelope(1, session_stdio(), sample_initialize());
        assert_eq!(serde_json::to_string(&event).unwrap(), GOLDEN_INITIALIZE);
    }

    #[test]
    fn golden_audit_gap_json() {
        let event = envelope(9, session_stdio(), sample_gap());
        assert_eq!(serde_json::to_string(&event).unwrap(), GOLDEN_AUDIT_GAP);
    }

    #[test]
    fn v1_tool_call_with_a_guard_is_rejected() {
        // The v2 break, tested rather than assumed: `suppressed_count` is
        // not a field this reader knows, and `deny_unknown_fields` must
        // refuse the line outright. Dropping that attribute, or adding an
        // `alias`/`default` that let a v1 line in, would read it with
        // `suppressed_other_count` silently zeroed — a wrong number where
        // the old one was merely coarse.
        let err = serde_json::from_str::<AuditEvent>(GOLDEN_TOOL_CALL_V1)
            .expect_err("a v1 tool_call carrying a guard must not parse under v2");
        assert!(
            err.to_string().contains("suppressed_count"),
            "the rejection must name the removed field: {err}"
        );
    }

    #[test]
    fn v1_lines_outside_the_break_still_parse_as_v1() {
        // The scope of that break, pinned on every line the docs promise
        // survives it: a v1 line naming no removed field still parses,
        // and still reads `v: 1`. The reader does not dispatch on `v`, so
        // a later "helpful" version check would widen the break past what
        // the schema documents — this test makes that a deliberate edit.
        let parsed: Vec<AuditEvent> = V1_LINES_OUTSIDE_THE_BREAK
            .iter()
            .map(|line| {
                serde_json::from_str(line).unwrap_or_else(|e| {
                    panic!("a v1 line outside the break stays readable: {line}: {e}")
                })
            })
            .collect();
        for event in &parsed {
            assert_eq!(event.v, 1, "the version is read back, never rewritten");
        }
        let AuditEventKind::ToolCall(tc) = &parsed[0].kind else {
            panic!("expected a tool_call event");
        };
        assert_eq!(tc.guard, None, "the guard-less shape is what survives");
        assert_eq!(tc.request.tool, "bug_url");
        assert!(matches!(parsed[1].kind, AuditEventKind::Initialize(_)));
        assert!(matches!(parsed[2].kind, AuditEventKind::AuditGap(_)));
    }

    #[test]
    fn reserved_identity_slots_are_absent_when_none() {
        // Both server-verified slots are `None` in every record this
        // build writes, and neither may appear as a key. Dropping either
        // `skip_serializing_if` would emit `"principal":null` /
        // `"work_context":null` — a reader cannot tell a null reserved
        // slot from a verified empty one.
        let line = serde_json::to_string(&envelope(1, session_stdio(), sample_initialize()))
            .expect("the fixture serializes");
        for key in ["principal", "work_context"] {
            assert!(
                !line.contains(key),
                "a reserved slot holding None must emit no key at all: {line}"
            );
        }
    }

    #[test]
    fn a_written_line_carries_the_current_schema_version() {
        // The goldens stamp `v` from the fixture; only `write_event`
        // stamps it on a real record, so this is the one test that reaches
        // that line. Asserted as the LITERAL bytes: comparing against
        // `SCHEMA_VERSION` would pass under any value of it.
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let sink = AuditSink::open(cfg.clone()).unwrap();
        sink.record(sample_tool_call(), session_http()).unwrap();
        let contents = std::fs::read_to_string(live_path(&cfg)).unwrap();
        assert!(
            contents.starts_with(r#"{"v":2,"#),
            "a written record must be stamped v2: {contents}"
        );
    }

    #[test]
    fn events_round_trip() {
        for event in [
            envelope(7, session_http(), sample_tool_call()),
            envelope(1, session_stdio(), sample_initialize()),
            envelope(9, session_stdio(), sample_gap()),
        ] {
            let s = serde_json::to_string(&event).unwrap();
            let back: AuditEvent = serde_json::from_str(&s).unwrap();
            assert_eq!(back, event);
        }
    }

    // ---------- traceparent parsing ----------

    /// The canonical W3C example, 55 bytes; the ids are the same as the
    /// golden serialization fixture's.
    const CANONICAL_TRACEPARENT: &str = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01";

    #[test]
    fn traceparent_canonical_value_parses_into_its_ids() {
        let trace = TraceContext::from_traceparent(CANONICAL_TRACEPARENT)
            .expect("the canonical W3C example must parse");
        assert_eq!(trace.trace_id, "0af7651916cd43dd8448eb211c80319c");
        assert_eq!(trace.span_id, "b7ad6b7169203331");
    }

    #[test]
    fn traceparent_flags_00_parses_and_flags_are_not_stored() {
        // The flags value is irrelevant (validated as hex, never stored):
        // an unsampled request correlates the same as a sampled one.
        let unsampled = CANONICAL_TRACEPARENT.replace("-01", "-00");
        let trace = TraceContext::from_traceparent(&unsampled).expect("flags 00 must parse");
        assert_eq!(trace.trace_id, "0af7651916cd43dd8448eb211c80319c");
        assert_eq!(trace.span_id, "b7ad6b7169203331");
    }

    #[test]
    fn traceparent_rejects_wrong_lengths() {
        assert_eq!(TraceContext::from_traceparent(""), None);
        // 54 bytes: the canonical value minus its last byte.
        let short = &CANONICAL_TRACEPARENT[..54];
        assert_eq!(TraceContext::from_traceparent(short), None);
        // 56 bytes: the canonical value plus one hex byte. A truncating
        // `>= 55` length check would parse this — the exact-length check
        // must not.
        let long = format!("{CANONICAL_TRACEPARENT}0");
        assert_eq!(TraceContext::from_traceparent(&long), None);
        // Oversized input is rejected on length alone (~1 MiB).
        let huge = "a".repeat(1 << 20);
        assert_eq!(TraceContext::from_traceparent(&huge), None);
    }

    #[test]
    fn traceparent_rejects_uppercase_hex() {
        // Uppercase in the trace-id: invalid per W3C.
        let upper_trace = "00-0AF7651916CD43DD8448EB211C80319C-b7ad6b7169203331-01";
        assert_eq!(TraceContext::from_traceparent(upper_trace), None);
        // Uppercase in the span-id.
        let upper_span = "00-0af7651916cd43dd8448eb211c80319c-B7AD6B7169203331-01";
        assert_eq!(TraceContext::from_traceparent(upper_span), None);
        // Uppercase hex in the version position.
        let upper_version = "0A-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01";
        assert_eq!(TraceContext::from_traceparent(upper_version), None);
    }

    #[test]
    fn traceparent_rejects_versions_other_than_00() {
        // `ff` is forbidden by W3C.
        let ff = "ff-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01";
        assert_eq!(TraceContext::from_traceparent(ff), None);
        // A future version is rejected even when it happens to be 55
        // bytes of version-00 shape: strict, no forward parsing.
        let v01 = "01-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01";
        assert_eq!(TraceContext::from_traceparent(v01), None);
        // The forward-format case W3C would allow lenient parsers to
        // accept: a future version with trailing data. Deliberately None.
        let forward = "01-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01-extra";
        assert_eq!(TraceContext::from_traceparent(forward), None);
    }

    #[test]
    fn traceparent_rejects_all_zero_ids() {
        let zero_trace = "00-00000000000000000000000000000000-b7ad6b7169203331-01";
        assert_eq!(TraceContext::from_traceparent(zero_trace), None);
        let zero_span = "00-0af7651916cd43dd8448eb211c80319c-0000000000000000-01";
        assert_eq!(TraceContext::from_traceparent(zero_span), None);
    }

    #[test]
    fn traceparent_rejects_non_hex_in_every_field() {
        for bad in [
            // `g` in the version, trace-id, span-id, and flags.
            "g0-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
            "00-gaf7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
            "00-0af7651916cd43dd8448eb211c80319c-g7ad6b7169203331-01",
            "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-g1",
            // A space in each field.
            " 0-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
            "00- af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
            "00-0af7651916cd43dd8448eb211c80319c- 7ad6b7169203331-01",
            "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331- 1",
        ] {
            assert_eq!(
                TraceContext::from_traceparent(bad),
                None,
                "must reject {bad:?}"
            );
        }
    }

    #[test]
    fn traceparent_rejects_misplaced_separators() {
        // Still 55 bytes each, but a `-` is off its fixed position.
        for bad in [
            // The trace-id/span-id separator shifted right by one.
            "00-0af7651916cd43dd8448eb211c80319cb-7ad6b7169203331-01",
            // The version separator replaced (two ids run together).
            "0000af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
            // The flags separator replaced by a hex byte.
            "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331001",
        ] {
            assert_eq!(bad.len(), 55, "fixture must stay 55 bytes");
            assert_eq!(
                TraceContext::from_traceparent(bad),
                None,
                "must reject {bad:?}"
            );
        }
    }

    #[test]
    fn traceparent_rejects_surrounding_whitespace() {
        // No trimming anywhere: whitespace is rejected, either by the
        // hex/layout checks (55 bytes) or by the length check (56).
        let lead_55 = format!(" {}", &CANONICAL_TRACEPARENT[..54]);
        assert_eq!(TraceContext::from_traceparent(&lead_55), None);
        let lead_56 = format!(" {CANONICAL_TRACEPARENT}");
        assert_eq!(TraceContext::from_traceparent(&lead_56), None);
        let trail_56 = format!("{CANONICAL_TRACEPARENT}\n");
        assert_eq!(TraceContext::from_traceparent(&trail_56), None);
    }

    #[test]
    fn traceparent_multibyte_unicode_returns_none_without_panicking() {
        // 53 ASCII bytes of the canonical value plus one 2-byte char:
        // exactly 55 bytes, so it passes the length check and must fall
        // out of the layout checks — never panic.
        let multibyte = format!("{}\u{e9}", &CANONICAL_TRACEPARENT[..53]);
        assert_eq!(multibyte.len(), 55, "fixture must be 55 bytes");
        assert_eq!(TraceContext::from_traceparent(&multibyte), None);
    }

    #[test]
    fn closed_vocabularies_have_stable_wire_names() {
        assert_eq!(json!(TransportKind::Stdio), json!("stdio"));
        assert_eq!(json!(TransportKind::Http), json!("http"));
        assert_eq!(json!(Verdict::Served), json!("served"));
        assert_eq!(json!(Verdict::ServedFiltered), json!("served_filtered"));
        assert_eq!(json!(Verdict::Denied), json!("denied"));
        assert_eq!(json!(Verdict::Refused), json!("refused"));
        assert_eq!(json!(OutcomeClass::Ok), json!("ok"));
        assert_eq!(json!(OutcomeClass::Refused), json!("refused"));
        assert_eq!(json!(OutcomeClass::Error), json!("error"));
        assert_eq!(json!(GapReason::WriteError), json!("write_error"));
        assert_eq!(json!(GapReason::RotationError), json!("rotation_error"));
    }

    #[test]
    fn unknown_envelope_key_rejected() {
        // An unknown top-level key flows into the flattened kind, whose
        // structs are closed — so the whole record is rejected.
        let with_extra = GOLDEN_AUDIT_GAP.replace("\"dropped\"", "\"bogus\":1,\"dropped\"");
        assert!(serde_json::from_str::<AuditEvent>(&with_extra).is_err());
    }

    // ---------- AuditConfig ----------

    #[test]
    fn config_minimal_file_applies_defaults() {
        let cfg = AuditConfig::from_toml_str("path = \"/var/log/bugwarden/audit.jsonl\"").unwrap();
        assert_eq!(
            cfg.path.as_deref(),
            Some(Path::new("/var/log/bugwarden/audit.jsonl"))
        );
        assert!(!cfg.fsync);
        assert_eq!(cfg.fail_mode, None);
        assert_eq!(cfg.rotate_max_bytes, 64 * 1024 * 1024);
        assert_eq!(cfg.rotate_keep, 8);
        assert!(cfg.suppressed_ids);
    }

    #[test]
    fn config_unknown_key_rejected_by_name() {
        let err = AuditConfig::from_toml_str("path = \"/a\"\nfrobnicate = true\n").unwrap_err();
        let msg = format!("{err:#}");
        assert!(msg.contains("frobnicate"), "unexpected error: {msg}");
    }

    #[test]
    fn config_rotate_keep_zero_with_rotation_rejected() {
        let err = AuditConfig::from_toml_str("path = \"/a\"\nrotate_keep = 0\n").unwrap_err();
        let msg = format!("{err:#}");
        assert!(msg.contains("rotate_keep"), "unexpected error: {msg}");
    }

    #[test]
    fn config_rotate_keep_zero_without_rotation_accepted() {
        let cfg =
            AuditConfig::from_toml_str("path = \"/a\"\nrotate_max_bytes = 0\nrotate_keep = 0\n")
                .unwrap();
        assert_eq!(cfg.rotate_max_bytes, 0);
        assert_eq!(cfg.rotate_keep, 0);
    }

    #[test]
    fn config_rotate_keep_above_cap_rejected() {
        let err = AuditConfig::from_toml_str("path = \"/a\"\nrotate_keep = 10001\n").unwrap_err();
        let msg = format!("{err:#}");
        assert!(
            msg.contains("rotate_keep") && msg.contains("10000"),
            "unexpected error: {msg}"
        );
    }

    #[test]
    fn config_rotate_keep_at_cap_accepted() {
        let cfg = AuditConfig::from_toml_str("path = \"/a\"\nrotate_keep = 10000\n").unwrap();
        assert_eq!(cfg.rotate_keep, 10_000);
    }

    #[test]
    fn config_every_fail_mode_parses() {
        for (raw, want) in [
            ("open", FailMode::Open),
            ("closed_writes_denials", FailMode::ClosedWritesDenials),
            ("closed_all", FailMode::ClosedAll),
        ] {
            let cfg =
                AuditConfig::from_toml_str(&format!("path = \"/a\"\nfail_mode = \"{raw}\"\n"))
                    .unwrap();
            assert_eq!(cfg.fail_mode, Some(want), "fail_mode = {raw}");
        }
        assert!(AuditConfig::from_toml_str("path = \"/a\"\nfail_mode = \"closed\"\n").is_err());
    }

    #[test]
    fn config_missing_path_rejected() {
        let err = AuditConfig::from_toml_str("fsync = true\n").unwrap_err();
        let msg = format!("{err:#}");
        assert!(msg.contains("path"), "unexpected error: {msg}");
    }

    #[test]
    fn shipped_example_audit_config_parses() {
        // The example in the repository must always load through the same
        // strict path operators use. Read at runtime (not include_str!) so
        // the crate stays packageable without the file.
        let path =
            std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../examples/audit.toml");
        let cfg = AuditConfig::load(&path).expect("example audit config must exist and parse");
        // The example spells out the defaults it documents; pin them so
        // the file and the code cannot drift apart silently.
        assert!(!cfg.fsync);
        assert_eq!(cfg.fail_mode, None);
        assert_eq!(cfg.rotate_max_bytes, 64 * 1024 * 1024);
        assert_eq!(cfg.rotate_keep, 8);
        assert!(cfg.suppressed_ids);
    }

    // ---------- AuditSink ----------

    #[test]
    #[cfg(unix)]
    fn sink_creates_file_0600_and_parent_0700() {
        use std::os::unix::fs::PermissionsExt as _;
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("sub/audit.jsonl");
        let sink = AuditSink::open(test_cfg(path.clone())).unwrap();
        sink.record(sample_initialize(), session_stdio()).unwrap();
        let file_mode = std::fs::metadata(&path).unwrap().permissions().mode();
        assert_eq!(file_mode & 0o777, 0o600, "audit file must be 0600");
        let dir_mode = std::fs::metadata(dir.path().join("sub"))
            .unwrap()
            .permissions()
            .mode();
        assert_eq!(dir_mode & 0o777, 0o700, "audit directory must be 0700");
    }

    #[test]
    fn sink_appends_across_open_calls_and_seq_restarts() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let first = AuditSink::open(cfg.clone()).unwrap();
        assert_eq!(
            first.record(sample_initialize(), session_stdio()).unwrap(),
            1
        );
        assert_eq!(
            first.record(sample_initialize(), session_stdio()).unwrap(),
            2
        );
        drop(first);
        // seq is per-process (per sink), not global: a fresh sink starts
        // at 1 again, but appends — it never truncates what came before.
        let second = AuditSink::open(cfg.clone()).unwrap();
        assert_eq!(
            second.record(sample_initialize(), session_stdio()).unwrap(),
            1
        );
        let seqs: Vec<u64> = read_events(live_path(&cfg)).iter().map(|e| e.seq).collect();
        assert_eq!(seqs, vec![1, 2, 1]);
    }

    #[test]
    #[cfg(unix)]
    fn sink_refuses_symlink_path() {
        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("target.jsonl");
        std::fs::write(&target, "").unwrap();
        let link = dir.path().join("link.jsonl");
        std::os::unix::fs::symlink(&target, &link).unwrap();
        let err = AuditSink::open(test_cfg(link)).unwrap_err();
        assert!(format!("{err:#}").contains("symlink"));
    }

    #[test]
    fn rotation_shifts_files_and_loses_no_event() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = test_cfg(dir.path().join("audit.jsonl"));
        // Roughly two records per file: multiple rotations over 8 records,
        // while rotate_keep is large enough that nothing gets deleted —
        // so every record written must still be on disk somewhere.
        cfg.rotate_max_bytes = 400;
        cfg.rotate_keep = 5;
        let sink = AuditSink::open(cfg.clone()).unwrap();
        for _ in 0..8 {
            sink.record(sample_initialize(), session_stdio()).unwrap();
        }
        // Oldest first: highest rotated number down to the live file.
        let mut seqs: Vec<u64> = Vec::new();
        let mut rotated = 0;
        for n in (1..=cfg.rotate_keep).rev() {
            let p = rotated_path(live_path(&cfg), n);
            if p.exists() {
                rotated += 1;
                seqs.extend(read_events(&p).iter().map(|e| e.seq));
            }
        }
        seqs.extend(read_events(live_path(&cfg)).iter().map(|e| e.seq));
        assert!(rotated >= 2, "expected multiple rotations, saw {rotated}");
        assert!(
            rotated <= cfg.rotate_keep,
            "rotated files exceed rotate_keep"
        );
        // No event lost, and seq strictly increasing across every
        // rotation boundary.
        assert_eq!(seqs, (1..=8).collect::<Vec<u64>>());
        // Rotation happens BEFORE the write, so no file — live or
        // rotated — ever exceeds the configured limit.
        assert_files_within_rotate_limit(&cfg);
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt as _;
            let mode = std::fs::metadata(live_path(&cfg))
                .unwrap()
                .permissions()
                .mode();
            assert_eq!(mode & 0o777, 0o600, "recreated live file must be 0600");
        }
    }

    #[test]
    fn rotation_deletes_files_beyond_rotate_keep() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = test_cfg(dir.path().join("audit.jsonl"));
        cfg.rotate_max_bytes = 400;
        cfg.rotate_keep = 1;
        let sink = AuditSink::open(cfg.clone()).unwrap();
        for _ in 0..8 {
            sink.record(sample_initialize(), session_stdio()).unwrap();
        }
        assert!(rotated_path(live_path(&cfg), 1).exists());
        assert!(
            !rotated_path(live_path(&cfg), 2).exists(),
            "rotation must not keep more than rotate_keep files"
        );
        // What survives still parses and stays in order (deletion loses
        // the oldest records — that is the operator's configured trade).
        let mut events = read_events(&rotated_path(live_path(&cfg), 1));
        events.extend(read_events(live_path(&cfg)));
        let seqs: Vec<u64> = events.iter().map(|e| e.seq).collect();
        assert!(
            seqs.len() < 8,
            "deletion beyond rotate_keep must have happened"
        );
        assert!(
            seqs.windows(2).all(|w| w[1] == w[0] + 1),
            "seqs stay contiguous: {seqs:?}"
        );
        assert_eq!(seqs.last(), Some(&8));
        // The size limit holds for what survives, too: rotation happens
        // before the write.
        assert_files_within_rotate_limit(&cfg);
    }

    #[test]
    fn single_oversized_record_written_alone_then_rotated_out() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = test_cfg(dir.path().join("audit.jsonl"));
        // Smaller than one serialized record: every record is oversized.
        cfg.rotate_max_bytes = 100;
        cfg.rotate_keep = 3;
        let sink = AuditSink::open(cfg.clone()).unwrap();
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            1
        );
        // The oversized record went into the empty live file as-is —
        // the one documented exception to the size bound.
        assert!(
            std::fs::metadata(live_path(&cfg)).unwrap().len() > cfg.rotate_max_bytes,
            "a single oversized record must be written despite the limit"
        );
        assert!(!rotated_path(live_path(&cfg), 1).exists());
        // The next record must rotate the oversized live file out first.
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            2
        );
        let rotated = read_events(&rotated_path(live_path(&cfg), 1));
        assert_eq!(rotated.len(), 1, "rotated file holds the first record");
        assert_eq!(rotated[0].seq, 1);
        let live = read_events(live_path(&cfg));
        assert_eq!(live.len(), 1, "live file holds only the new record");
        assert_eq!(live[0].seq, 2);
    }

    #[test]
    fn reopened_sink_counts_existing_bytes_toward_rotation() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = test_cfg(dir.path().join("audit.jsonl"));
        // One record nearly fills the limit; the next must rotate — but
        // only if the reopened sink counted the bytes already on disk.
        cfg.rotate_max_bytes = 200;
        cfg.rotate_keep = 3;
        let first = AuditSink::open(cfg.clone()).unwrap();
        first.record(sample_initialize(), session_stdio()).unwrap();
        drop(first);
        let second = AuditSink::open(cfg.clone()).unwrap();
        second.record(sample_initialize(), session_stdio()).unwrap();
        assert!(
            rotated_path(live_path(&cfg), 1).exists(),
            "reopen must count existing bytes toward the rotation limit"
        );
        let live = read_events(live_path(&cfg));
        assert_eq!(live.len(), 1, "live file holds only the new record");
        assert_eq!(live[0].seq, 1, "a fresh sink assigns seq from 1");
        assert_eq!(read_events(&rotated_path(live_path(&cfg), 1)).len(), 1);
    }

    #[test]
    fn concurrent_records_are_contiguous_and_in_file_order() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let sink = AuditSink::open(cfg.clone()).unwrap();
        std::thread::scope(|s| {
            for _ in 0..4 {
                s.spawn(|| {
                    for _ in 0..25 {
                        sink.record(sample_initialize(), session_stdio())
                            .expect("concurrent record must succeed");
                    }
                });
            }
        });
        let seqs: Vec<u64> = read_events(live_path(&cfg)).iter().map(|e| e.seq).collect();
        // Unique, contiguous from 1, and file order equals seq order —
        // all three in one assertion.
        assert_eq!(seqs, (1..=100).collect::<Vec<u64>>());
    }

    #[test]
    fn gap_marker_precedes_next_record_after_write_failures() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let sink = AuditSink::open(cfg.clone()).unwrap();
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            1
        );
        sink.set_fail_writes(true);
        for _ in 0..2 {
            let err = sink
                .record(sample_initialize(), session_stdio())
                .unwrap_err();
            assert!(matches!(err, AuditError::Write(_)));
        }
        sink.set_fail_writes(false);
        // The next successful record is preceded by an audit_gap carrying
        // the dropped count; the gap consumes a seq of its own.
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            3
        );
        let events = read_events(live_path(&cfg));
        assert_eq!(events.len(), 3);
        assert_eq!(events[1].seq, 2);
        match &events[1].kind {
            AuditEventKind::AuditGap(gap) => {
                assert_eq!(gap.dropped, 2);
                assert_eq!(gap.reason, GapReason::WriteError);
            }
            other => panic!("expected audit_gap, got {other:?}"),
        }
        assert_eq!(events[2].seq, 3);
    }

    #[test]
    fn gap_marker_after_rotation_failure_reports_rotation_error() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = test_cfg(dir.path().join("audit.jsonl"));
        // One record nearly fills the limit, so every following write
        // wants a rotation first.
        cfg.rotate_max_bytes = 200;
        cfg.rotate_keep = 4;
        let sink = AuditSink::open(cfg.clone()).unwrap();
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            1
        );
        sink.set_fail_rotation(true);
        for _ in 0..2 {
            let err = sink
                .record(sample_initialize(), session_stdio())
                .unwrap_err();
            assert!(matches!(err, AuditError::Rotation(_)));
        }
        sink.set_fail_rotation(false);
        // Recovery: the gap record precedes the next record. Each write
        // may itself rotate, so collect across all generations.
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            3
        );
        let mut raw = String::new();
        let mut events = Vec::new();
        for n in (1..=cfg.rotate_keep).rev() {
            let p = rotated_path(live_path(&cfg), n);
            if p.exists() {
                raw.push_str(&std::fs::read_to_string(&p).unwrap());
                events.extend(read_events(&p));
            }
        }
        raw.push_str(&std::fs::read_to_string(live_path(&cfg)).unwrap());
        events.extend(read_events(live_path(&cfg)));
        let seqs: Vec<u64> = events.iter().map(|e| e.seq).collect();
        assert_eq!(seqs, vec![1, 2, 3]);
        match &events[1].kind {
            AuditEventKind::AuditGap(gap) => {
                assert_eq!(gap.dropped, 2);
                assert_eq!(gap.reason, GapReason::RotationError);
            }
            other => panic!("expected audit_gap, got {other:?}"),
        }
        assert!(
            raw.contains(r#""reason":"rotation_error""#),
            "the gap record on disk must name the rotation failure"
        );
    }

    #[test]
    fn partial_write_leaves_torn_line_and_stream_self_heals() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let sink = AuditSink::open(cfg.clone()).unwrap();
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            1
        );
        sink.set_fail_writes_partial(true);
        let err = sink
            .record(sample_initialize(), session_stdio())
            .unwrap_err();
        assert!(matches!(err, AuditError::Write(_)));
        sink.set_fail_writes_partial(false);
        let raw = std::fs::read_to_string(live_path(&cfg)).unwrap();
        assert!(
            !raw.ends_with('\n'),
            "the partial write must have left a torn, unterminated line"
        );
        // Recovery: the healing newline terminates the torn line; the
        // gap record and the caller's record follow, both parsable.
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            3
        );
        let raw = std::fs::read_to_string(live_path(&cfg)).unwrap();
        let mut torn_lines = 0;
        let mut events: Vec<AuditEvent> = Vec::new();
        for line in raw.lines().filter(|l| !l.is_empty()) {
            match serde_json::from_str(line) {
                Ok(event) => events.push(event),
                Err(_) => torn_lines += 1,
            }
        }
        assert_eq!(torn_lines, 1, "at most one torn line per outage");
        let seqs: Vec<u64> = events.iter().map(|e| e.seq).collect();
        assert_eq!(seqs, vec![1, 2, 3], "complete records stay contiguous");
        match &events[1].kind {
            AuditEventKind::AuditGap(gap) => {
                assert_eq!(gap.dropped, 1);
                assert_eq!(gap.reason, GapReason::WriteError);
            }
            other => panic!("expected audit_gap, got {other:?}"),
        }
    }

    #[test]
    fn real_sink_stamps_ts_rfc3339_utc_millis() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let sink = AuditSink::open(cfg.clone()).unwrap();
        sink.record(sample_initialize(), session_stdio()).unwrap();
        let events = read_events(live_path(&cfg));
        let ts = &events[0].ts;
        // ^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$, spelled out
        // positionally ('d' = ASCII digit) to avoid a regex dependency.
        const PATTERN: &[u8] = b"dddd-dd-ddTdd:dd:dd.dddZ";
        assert_eq!(ts.len(), PATTERN.len(), "unexpected ts length: {ts:?}");
        for (i, (&got, &want)) in ts.as_bytes().iter().zip(PATTERN).enumerate() {
            match want {
                b'd' => assert!(got.is_ascii_digit(), "byte {i} of ts {ts:?} not a digit"),
                _ => assert_eq!(got, want, "byte {i} of ts {ts:?}"),
            }
        }
    }

    #[test]
    fn api_key_canary_never_reaches_the_file() {
        // Regression tripwire: the schema has no field that could carry
        // the Bugzilla API key, and the sink writes only what the schema
        // holds. Write every record kind with every field populated and
        // grep the file for a secret-shaped canary. The canary is a
        // local constant — nothing in this module reads the environment,
        // and mutating the process environment would race the parallel
        // test harness.
        const CANARY: &str = "bugwarden-canary-3f6d1a-api-key-value";
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let sink = AuditSink::open(cfg.clone()).unwrap();
        sink.record(sample_tool_call(), session_http()).unwrap();
        sink.record(sample_initialize(), session_stdio()).unwrap();
        sink.record(sample_gap(), session_stdio()).unwrap();
        let contents = std::fs::read_to_string(live_path(&cfg)).unwrap();
        assert_eq!(contents.lines().count(), 3);
        assert!(
            !contents.contains(CANARY),
            "the API key canary leaked into the audit file"
        );
    }

    // ---------- request-path wiring ----------

    #[test]
    fn fail_mode_transport_defaults_bind_the_documented_wording() {
        // The documented derivation: stdio (a local, single-operator
        // session) defaults to open — availability for a single user;
        // http (remote clients) defaults to closed_all — accountability
        // for a fleet.
        assert_eq!(FailMode::default_for(TransportKind::Stdio), FailMode::Open);
        assert_eq!(
            FailMode::default_for(TransportKind::Http),
            FailMode::ClosedAll
        );
        // An explicit configured value wins over both derivations.
        for transport in [TransportKind::Stdio, TransportKind::Http] {
            assert_eq!(
                FailMode::resolve(Some(FailMode::ClosedWritesDenials), transport),
                FailMode::ClosedWritesDenials
            );
        }
        assert_eq!(
            FailMode::resolve(None, TransportKind::Stdio),
            FailMode::Open
        );
        assert_eq!(
            FailMode::resolve(None, TransportKind::Http),
            FailMode::ClosedAll
        );
    }

    #[test]
    fn audit_state_session_id_is_pid_dash_epoch() {
        let dir = tempfile::tempdir().unwrap();
        let sink = AuditSink::open(test_cfg(dir.path().join("audit.jsonl"))).unwrap();
        let state = AuditState::new(sink, FailMode::Open, None);
        let (pid, epoch) = state
            .stdio_session_id
            .split_once('-')
            .expect("session id is <pid>-<epoch>");
        assert_eq!(pid, std::process::id().to_string());
        assert!(epoch.parse::<u64>().is_ok(), "epoch part is seconds");
    }

    #[test]
    fn cell_verdict_merge_keeps_the_worst_and_its_rule() {
        let cell = AuditCell::default();
        cell.note_verdict_rule(Verdict::Served, "allow-most");
        cell.note_verdict_rule(Verdict::Denied, "embargo");
        // A lesser verdict after the worst changes nothing.
        cell.note_verdict_rule(Verdict::ServedFiltered, "peek");
        let info = cell.into_guard_info(None, true).expect("verdict noted");
        assert_eq!(info.verdict, Verdict::Denied);
        assert_eq!(info.rule.as_deref(), Some("embargo"));
    }

    #[test]
    fn cell_upgrade_by_a_rule_less_verdict_clears_the_rule() {
        // The rule names the DECIDING rule of the worst verdict: when a
        // denial no single rule decided outranks a rule-decided serve, no
        // rule may stay attached. A policy-DEFAULT denial is not this
        // case — it decides under the name "default" and arrives through
        // note_verdict_rule.
        let cell = AuditCell::default();
        cell.note_verdict_rule(Verdict::Served, "allow-most");
        cell.note_verdict(Verdict::Denied);
        let info = cell.into_guard_info(None, true).expect("verdict noted");
        assert_eq!(info.verdict, Verdict::Denied);
        assert_eq!(info.rule, None);
    }

    #[test]
    fn cell_first_rule_wins_at_equal_severity() {
        let cell = AuditCell::default();
        cell.note_verdict(Verdict::Served);
        cell.note_verdict_rule(Verdict::Served, "first");
        cell.note_verdict_rule(Verdict::Served, "second");
        let info = cell.into_guard_info(None, true).expect("verdict noted");
        assert_eq!(info.rule.as_deref(), Some("first"));
    }

    #[test]
    fn cell_suppression_implies_a_filtered_serve() {
        let cell = AuditCell::default();
        cell.note_verdict_rule(Verdict::Served, "allow-most");
        cell.note_suppressed([1290040, 1290041, 1290040]);
        let info = cell.into_guard_info(Some("sha256:ab"), true).unwrap();
        assert_eq!(info.verdict, Verdict::ServedFiltered);
        assert_eq!(info.suppressed_ids, vec![1290040, 1290041]);
        assert_eq!(info.suppressed_ids_count, 2);
        assert_eq!(info.suppressed_other_count, 0);
        assert_eq!(info.policy_hash.as_deref(), Some("sha256:ab"));
    }

    #[test]
    fn cell_split_counters_report_their_own_populations() {
        // Issue #68 at v2: each tally ships as itself. The two numbers
        // are deliberately DIFFERENT, so swapping the fields fails here —
        // which a single summed field structurally could not catch. The
        // id-less notes also accumulate rather than last-note-wins.
        let cell = AuditCell::default();
        cell.note_suppressed([7]);
        cell.note_suppressed_count(2);
        cell.note_suppressed_count(1);
        let info = cell.into_guard_info(None, true).unwrap();
        assert_eq!(info.suppressed_ids_count, 1, "one named id");
        assert_eq!(
            info.suppressed_other_count, 3,
            "two id-less suppressions plus one more, added not replaced"
        );
        assert_eq!(info.suppressed_ids, vec![7]);
    }

    #[test]
    fn cell_split_survives_the_ids_config() {
        // With the ids elided the emitted vector is empty, so a count
        // derived from IT would read zero — and the operator's only
        // remaining signal that two bugs were withheld would vanish.
        let cell = AuditCell::default();
        cell.note_suppressed([40, 41]);
        cell.note_suppressed_count(3);
        let info = cell.into_guard_info(None, false).unwrap();
        assert!(info.suppressed_ids.is_empty(), "ids elided by config");
        assert_eq!(
            info.suppressed_ids_count, 2,
            "counted from the set, not from the vector the config emptied"
        );
        assert_eq!(
            info.suppressed_other_count, 3,
            "the id-less tally is untouched"
        );
    }

    #[test]
    fn cell_untouched_yields_no_guard_info() {
        let cell = AuditCell::default();
        assert_eq!(cell.into_guard_info(Some("sha256:ab"), true), None);
        // Upstream info alone is not a guard verdict. It is taken FIRST:
        // into_guard_info drains the whole cell.
        let cell = AuditCell::default();
        cell.note_upstream(UpstreamInfo {
            requests: 1,
            status: Some(200),
            latency_ms: 3,
        });
        assert!(cell.take_upstream().is_some());
        assert_eq!(cell.into_guard_info(None, true), None);
    }

    #[test]
    fn cell_redaction_is_recorded_and_upgrades_the_verdict() {
        let cell = AuditCell::default();
        cell.note_verdict(Verdict::Served);
        cell.note_redacted("summary_view");
        cell.note_redacted("summary_view");
        let info = cell.into_guard_info(None, true).unwrap();
        assert_eq!(info.verdict, Verdict::ServedFiltered);
        assert_eq!(info.redacted_fields, vec!["summary_view".to_string()]);
    }

    #[test]
    fn cell_client_window_redaction_keeps_the_granting_rule() {
        // Issue #67: a client-requested window is not a guard decision, so
        // it must not take the granting rule down with it — the two notes
        // are pinned side by side because they differ ONLY in that.
        let cell = AuditCell::default();
        cell.note_verdict_rule(Verdict::Served, "allow-desktop");
        cell.note_redacted_client_window("attachment_window");
        let info = cell.into_guard_info(None, true).unwrap();
        assert_eq!(info.verdict, Verdict::ServedFiltered);
        assert_eq!(
            info.redacted_fields,
            vec!["attachment_window".to_string()],
            "the window is still named"
        );
        assert_eq!(
            info.rule.as_deref(),
            Some("allow-desktop"),
            "the rule that granted the call survives the client's window"
        );

        let cell = AuditCell::default();
        cell.note_verdict_rule(Verdict::Served, "allow-desktop");
        cell.note_redacted("summary_view");
        let info = cell.into_guard_info(None, true).unwrap();
        assert_eq!(info.verdict, Verdict::ServedFiltered);
        assert_eq!(
            info.rule, None,
            "while a guard redaction that UPGRADES the verdict clears it"
        );

        // ...but only because it upgraded. Recorded at equal rank — the
        // shape bug_info actually produces — the ordinary merge fills an
        // empty slot only, so the rule stands and the two notes agree.
        let cell = AuditCell::default();
        cell.note_verdict_rule(Verdict::ServedFiltered, "summaries-only");
        cell.note_redacted("summary_view");
        let info = cell.into_guard_info(None, true).unwrap();
        assert_eq!(info.verdict, Verdict::ServedFiltered);
        assert_eq!(info.rule.as_deref(), Some("summaries-only"));

        // A guard note landing AFTER a client window merges by the
        // ordinary rules: equal rank here, so the window's rule stands.
        let cell = AuditCell::default();
        cell.note_verdict_rule(Verdict::Served, "allow-desktop");
        cell.note_redacted_client_window("comments_window");
        cell.note_redacted("summary_view");
        let info = cell.into_guard_info(None, true).unwrap();
        assert_eq!(info.rule.as_deref(), Some("allow-desktop"));
    }

    #[test]
    fn cell_scan_with_drops_is_a_filtered_serve() {
        let cell = AuditCell::default();
        cell.note_verdict(Verdict::Served);
        cell.note_scan(200, 3);
        let info = cell.into_guard_info(None, true).unwrap();
        assert_eq!(
            info.verdict,
            Verdict::ServedFiltered,
            "a search that withheld rows is a filtered serve"
        );
        assert_eq!(
            info.scan,
            Some(ScanInfo {
                scanned: 200,
                dropped: 3
            })
        );
    }

    #[test]
    fn cell_clean_scan_records_numbers_without_touching_the_verdict() {
        let cell = AuditCell::default();
        cell.note_verdict(Verdict::Served);
        cell.note_scan(30, 0);
        let info = cell.into_guard_info(None, true).unwrap();
        assert_eq!(info.verdict, Verdict::Served, "a clean scan stays served");
        assert_eq!(
            info.scan,
            Some(ScanInfo {
                scanned: 30,
                dropped: 0
            })
        );
    }

    #[test]
    fn cell_scan_noted_twice_keeps_the_last_note() {
        // One search per call by construction; should a second note ever
        // happen, last write wins — documented, not merged.
        let cell = AuditCell::default();
        cell.note_scan(200, 3);
        cell.note_scan(400, 0);
        let info = cell.into_guard_info(None, true).unwrap();
        assert_eq!(
            info.scan,
            Some(ScanInfo {
                scanned: 400,
                dropped: 0
            })
        );
        // The earlier note's verdict upgrade is NOT rolled back: rows
        // were withheld at some point of this call, so the record says
        // filtered (worst-wins, like every other verdict source).
        assert_eq!(info.verdict, Verdict::ServedFiltered);
    }

    #[test]
    fn cell_scan_counts_survive_the_suppressed_ids_config() {
        // `suppressed_ids = false` elides ids; the scan carries counts
        // only, so it ships untouched (acceptance (c) of issue #29).
        let cell = AuditCell::default();
        cell.note_scan(30, 2);
        cell.note_suppressed([12, 19]);
        let info = cell.into_guard_info(None, false).unwrap();
        assert!(info.suppressed_ids.is_empty(), "ids elided by config");
        assert_eq!(info.suppressed_ids_count, 2, "the count still ships");
        assert_eq!(info.suppressed_other_count, 0);
        assert_eq!(
            info.scan,
            Some(ScanInfo {
                scanned: 30,
                dropped: 2
            }),
            "the scan accounting is counts, never ids"
        );
    }

    #[test]
    fn policy_hash_of_digests_the_given_bytes() {
        // The FIPS 180-2 test vector for sha256("abc"), precomputed:
        // recomputing the expectation with sha2 here would only prove the
        // function calls sha2, not that it hashes the bytes it was given.
        assert_eq!(
            policy_hash_of(b"abc"),
            "sha256:ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
        );
    }

    // ---------- sink selection (issue #31, revised 2026-08-18) ----------

    #[test]
    fn select_sinks_resolves_the_four_deployments() {
        assert_eq!(select_sinks(None, false).unwrap(), SinkSelection::NoAudit);
        assert_eq!(
            select_sinks(Some(Path::new("/etc/bugwarden/audit.toml")), false).unwrap(),
            SinkSelection::FileOnly
        );
        assert_eq!(
            select_sinks(Some(Path::new("/etc/bugwarden/audit.toml")), true).unwrap(),
            SinkSelection::Both
        );
        assert_eq!(
            select_sinks(Some(Path::new(AUDIT_CONFIG_NONE)), true).unwrap(),
            SinkSelection::OtlpOnly
        );
    }

    #[test]
    fn select_sinks_refuses_the_two_ambiguous_spellings() {
        // `none` with no OTLP endpoint: an explicit request for a sink
        // that leaves zero of them.
        let err = format!(
            "{}",
            select_sinks(Some(Path::new("none")), false).unwrap_err()
        );
        assert!(
            err.contains("no audit sink") && err.contains("OTEL_EXPORTER_OTLP_ENDPOINT"),
            "the refusal must explain both ways out: {err}"
        );
        // An endpoint with no file decision: OTLP-only must be said, not
        // inferred, or a typo'd BUGWARDEN_AUDIT_CONFIG silently drops the
        // file from a deployment that meant to run both.
        let err = format!("{}", select_sinks(None, true).unwrap_err());
        assert!(
            err.contains("BUGWARDEN_AUDIT_CONFIG") && err.contains("`none`"),
            "the refusal must name the explicit spellings: {err}"
        );
    }

    #[test]
    fn the_none_sentinel_is_exact_bytes() {
        // Anything else is a path: `None` and `./none` are files an
        // operator can really have, and over-matching would disable
        // auditing on a spelling nobody wrote.
        assert_eq!(
            select_sinks(Some(Path::new("None")), true).unwrap(),
            SinkSelection::Both
        );
        assert_eq!(
            select_sinks(Some(Path::new("./none")), true).unwrap(),
            SinkSelection::Both
        );
    }

    // ---------- the exporter as a load-bearing sink ----------

    /// A scriptable exporter for sink-side tests.
    #[derive(Debug, Default)]
    struct StubExport {
        refuse: std::sync::atomic::AtomicBool,
        failing: std::sync::atomic::AtomicBool,
        lost: std::sync::atomic::AtomicU64,
        accepted: Mutex<Vec<Vec<u8>>>,
    }

    impl AuditExport for StubExport {
        fn accept(&self, _event: &AuditEvent, line: &[u8]) -> Result<(), ExportRefused> {
            if self.refuse.load(std::sync::atomic::Ordering::Relaxed) {
                return Err(ExportRefused);
            }
            self.accepted.lock().expect("stub lock").push(line.to_vec());
            Ok(())
        }

        fn delivery_failing(&self) -> bool {
            self.failing.load(std::sync::atomic::Ordering::Relaxed)
        }

        fn take_lost(&self) -> u64 {
            self.lost.swap(0, std::sync::atomic::Ordering::Relaxed)
        }
    }

    #[test]
    fn a_fileless_sink_records_through_the_exporter_alone() {
        let cfg = AuditConfig::fileless();
        assert!(cfg.path.is_none(), "fileless names no file");
        let export = std::sync::Arc::new(StubExport::default());
        let sink = AuditSink::open(cfg).unwrap().with_export(export.clone());
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            1
        );
        assert_eq!(sink.record(sample_tool_call(), session_http()).unwrap(), 2);
        let accepted = export.accepted.lock().expect("stub lock").clone();
        assert_eq!(accepted.len(), 2, "every record reaches the exporter");
        // The handed-over line is the record, parseable and in seq order —
        // the same bytes a file would have carried.
        let events: Vec<AuditEvent> = accepted
            .iter()
            .map(|line| serde_json::from_slice(line).expect("a parseable record line"))
            .collect();
        assert_eq!(events.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 2]);
        assert!(!sink.failing());
    }

    #[test]
    fn an_export_refusal_fails_the_record_and_the_file_still_keeps_it() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let export = std::sync::Arc::new(StubExport::default());
        let sink = AuditSink::open(cfg.clone())
            .unwrap()
            .with_export(export.clone());
        export
            .refuse
            .store(true, std::sync::atomic::Ordering::Relaxed);
        let err = sink
            .record(sample_initialize(), session_stdio())
            .unwrap_err();
        assert!(
            matches!(err, AuditError::Export),
            "a refused hand-off is an export failure: {err:?}"
        );
        // The FILE write preceded the refusal and stands, its seq spent:
        // the loss accounting over-reports rather than under-reports,
        // exactly as a failed fsync does.
        let events = read_events(live_path(&cfg));
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].seq, 1);
        assert!(sink.failing(), "a refused record puts the sink in failure");
        // Recovery: the next record is preceded by the gap accounting it.
        export
            .refuse
            .store(false, std::sync::atomic::Ordering::Relaxed);
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            3
        );
        let events = read_events(live_path(&cfg));
        assert_eq!(events.len(), 3);
        match &events[1].kind {
            AuditEventKind::AuditGap(gap) => {
                assert_eq!(gap.dropped, 1);
                assert_eq!(gap.reason, GapReason::WriteError);
            }
            other => panic!("expected audit_gap, got {other:?}"),
        }
    }

    #[test]
    fn export_delivery_failure_alone_puts_the_sink_in_failure() {
        // The gate condition: delivery health, not only a counter — a
        // counter clears the moment its gap is queued, and a gate read
        // off it alone would flap once per batch for the whole outage.
        let export = std::sync::Arc::new(StubExport::default());
        let sink = AuditSink::open(AuditConfig::fileless())
            .unwrap()
            .with_export(export.clone());
        assert!(!sink.failing());
        export
            .failing
            .store(true, std::sync::atomic::Ordering::Relaxed);
        assert!(
            sink.failing(),
            "failing delivery must gate even with nothing dropped"
        );
        export
            .failing
            .store(false, std::sync::atomic::Ordering::Relaxed);
        assert!(!sink.failing(), "recovered delivery must clear the gate");
    }

    #[test]
    fn undelivered_export_losses_surface_as_an_audit_gap() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let export = std::sync::Arc::new(StubExport::default());
        let sink = AuditSink::open(cfg.clone())
            .unwrap()
            .with_export(export.clone());
        // Three records accepted earlier were never delivered.
        export.lost.store(3, std::sync::atomic::Ordering::Relaxed);
        assert!(sink.failing(), "known losses put the sink in failure");
        assert_eq!(
            sink.record(sample_initialize(), session_stdio()).unwrap(),
            2
        );
        let events = read_events(live_path(&cfg));
        assert_eq!(events.len(), 2);
        match &events[0].kind {
            AuditEventKind::AuditGap(gap) => {
                assert_eq!(gap.dropped, 3, "the gap carries the loss count");
                assert_eq!(gap.reason, GapReason::WriteError);
            }
            other => panic!("expected audit_gap, got {other:?}"),
        }
        // The gap record itself is exported too: the collector sees its
        // own outage accounted in the stream it lost records from.
        let accepted = export.accepted.lock().expect("stub lock").clone();
        assert_eq!(accepted.len(), 2, "gap and record both reach the exporter");
    }

    /// An exporter that proves the file write PRECEDES the hand-off: at
    /// accept time the line must already be on disk.
    #[derive(Debug)]
    struct FileCheckingExport {
        path: PathBuf,
        checked: std::sync::atomic::AtomicU64,
    }

    impl AuditExport for FileCheckingExport {
        fn accept(&self, _event: &AuditEvent, line: &[u8]) -> Result<(), ExportRefused> {
            let contents = std::fs::read(&self.path).expect("the audit file exists at accept");
            assert!(
                contents.windows(line.len()).any(|window| window == line),
                "the file must already hold the record when the exporter sees it"
            );
            self.checked
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            Ok(())
        }

        fn delivery_failing(&self) -> bool {
            false
        }

        fn take_lost(&self) -> u64 {
            0
        }
    }

    #[test]
    fn with_both_sinks_the_file_write_precedes_the_export() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = test_cfg(dir.path().join("audit.jsonl"));
        let export = std::sync::Arc::new(FileCheckingExport {
            path: live_path(&cfg).to_path_buf(),
            checked: std::sync::atomic::AtomicU64::new(0),
        });
        let sink = AuditSink::open(cfg.clone())
            .unwrap()
            .with_export(export.clone());
        sink.record(sample_initialize(), session_stdio()).unwrap();
        sink.record(sample_tool_call(), session_http()).unwrap();
        assert_eq!(
            export.checked.load(std::sync::atomic::Ordering::Relaxed),
            2,
            "the ordering assertion must actually have run"
        );
    }

    #[test]
    fn policy_hash_of_is_fixed_width_lowercase_hex() {
        // The vector above pins one exact value; this pins the encoding
        // for inputs of every length up to a block and a bit. Of these 65
        // digests 55 carry a byte below 0x10 and all 65 carry one above
        // 0x9f, so a dropped zero pad shortens the string here and an
        // uppercase digit surfaces here — whichever input hits it.
        for len in 0..=64 {
            let hash = policy_hash_of(&vec![b'x'; len]);
            let hex = hash
                .strip_prefix("sha256:")
                .expect("every hash carries the algorithm prefix");
            assert_eq!(hex.len(), 64, "input of {len} bytes: {hash}");
            assert!(
                hex.bytes()
                    .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()),
                "input of {len} bytes: {hash}"
            );
        }
    }
}