ebman 0.37.0

//! Audit log: writers, parser, renderer, filter.
//!
//! Writers and parser are co-located so the line format has a single
//! source of truth. The typed writer APIs cover every shape:
//!
//! - `append_action_dispatched` / `append_action_completed` —
//!   normal TUI action lines (rebuild / restart / deploy / etc.).
//! - `append_action_skipped` / `append_action_undone` — one-shot
//!   action lines with no completion pair (a batch member skipped
//!   because its env left the view; an operator-driven undo).
//! - `append_rollout` — cross-region rollout lines tagged with a
//!   per-run `rollout_id` for post-mortem correlation.
//! - `append_lint_fix` — `ebman lint --fix` dispatches, tagged
//!   with the originating `rule_id`.
//! - `append_dlq_op` — one-shot DLQ ops (delete / resend / purge /
//!   replay), recorded at fire time.
//!
//! Plus `append_raw` — the lower-level "I already have a detail
//! string" entry point. As of 0.24 the hand-rolled `append_raw` action
//! sites have all moved to the typed siblings above; the only remaining
//! caller is the passive `stage=event kind=red_transition` health-log
//! line, which is genuinely an event, not an action.
//!
//! All paths funnel into the same private `write_audit_line`
//! helper (or its `_raw` sibling) so file rotation + webhook
//! fan-out apply uniformly to every line type.
//!
//! Line shapes:
//!
//! - Normal action:
//!   `{rfc3339}\taccount=A\tprofile=P\tregion=R\tstage=S action=Act target=Env [outcome=ok|err="..."]`
//! - Rollout:
//!   `{rfc3339}\trollout_id=ID\tprofile=P\tregion=R\tstage=S action=Rollout target=Env version=V [outcome=ok|err="..."]`
//!   — `profile` was INSERTED BEFORE `region` in 0.34.2, so a
//!   positional/tab-index reader of rollout lines shifts by one. Nothing
//!   advertised parses positionally (`ebman audit --json` is the
//!   supported consumer and `parse_kv_pairs` is key-value), but it is
//!   the sentence a log-shipper owner needs.
//! - Lint fix:
//!   `{rfc3339}\tregion=R\tstage=fix action=SetOption target=Env rule_id=ID namespace=NS name=N value="V" outcome=ok|err="..."`
//!
//! The parser handles all three shapes uniformly: split on tab, then
//! tokenize every chunk as `key=value` pairs (with quoted-value
//! support). Known keys get promoted into typed fields on
//! `AuditEntry`; unknown keys land in `extras` so we don't drop
//! information.
//!
//! [`ebman audit`](../bin/ebman/cli/audit/index.html) — the CLI — uses
//! `parse_audit_line` + `AuditFilter` + the render helpers below
//! to surface entries for scripting / Slack-bot routing / on-call
//! dashboards / CI gating.

use std::collections::BTreeMap;

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct AuditEntry {
    pub when: String,
    pub account: Option<String>,
    pub profile: Option<String>,
    pub region: Option<String>,
    pub rollout_id: Option<String>,
    pub stage: Option<String>,
    pub action: Option<String>,
    pub target: Option<String>,
    pub version: Option<String>,
    pub rule_id: Option<String>,
    pub outcome: Option<String>,
    pub err: Option<String>,
    pub extras: BTreeMap<String, String>,
    pub raw: String,
}

/// Parse one audit-log line. Returns `None` for blank lines or lines
/// without an RFC3339-shaped timestamp as the first tab field.
pub(crate) fn parse_audit_line(line: &str) -> Option<AuditEntry> {
    let line = line.trim_end_matches('\n').trim_end_matches('\r');
    if line.is_empty() {
        return None;
    }
    let mut tabs = line.splitn(2, '\t');
    let when = tabs.next()?.trim().to_string();
    // Sanity: timestamp should at least look RFC3339-ish (`YYYY-MM-DDT...`).
    if when.len() < 10 || when.chars().nth(4) != Some('-') {
        return None;
    }
    let rest = tabs.next().unwrap_or("");
    // Tokenize the whole rest-of-line as key=value pairs. Tab and
    // space both function as separators; the kv parser walks until
    // the next `key=` boundary regardless of which one delimits.
    let pairs = parse_kv_pairs(rest);

    let mut entry = AuditEntry {
        when,
        account: None,
        profile: None,
        region: None,
        rollout_id: None,
        stage: None,
        action: None,
        target: None,
        version: None,
        rule_id: None,
        outcome: None,
        err: None,
        extras: BTreeMap::new(),
        raw: line.to_string(),
    };
    for (k, v) in pairs {
        // Treat literal "-" as missing — same convention
        // `write_audit_line` uses for absent account / profile.
        let v_opt = if v == "-" { None } else { Some(v) };
        match k.as_str() {
            "account" => entry.account = v_opt,
            "profile" => entry.profile = v_opt,
            "region" => entry.region = v_opt,
            "rollout_id" => entry.rollout_id = v_opt,
            "stage" => entry.stage = v_opt,
            "action" => entry.action = v_opt,
            "target" => entry.target = v_opt,
            "version" => entry.version = v_opt,
            "rule_id" => entry.rule_id = v_opt,
            "outcome" => entry.outcome = v_opt,
            "err" => entry.err = v_opt,
            _ => {
                if let Some(v) = v_opt {
                    entry.extras.insert(k, v);
                }
            }
        }
    }
    Some(entry)
}

/// Tokenize a string into `key=value` pairs. Keys are `[A-Za-z0-9_]+`.
/// Values are either `"quoted"` (everything between matched `"`s) or
/// unquoted (everything from `=` to the next whitespace-then-key=
/// boundary or end-of-string). Naked spaces inside an unquoted value
/// are preserved — e.g. `target=env-a ↔ env-b stage=dispatched`
/// yields `target` = `"env-a ↔ env-b"` and `stage` = `"dispatched"`.
pub(crate) fn parse_kv_pairs(text: &str) -> Vec<(String, String)> {
    let chars: Vec<char> = text.chars().collect();
    let n = chars.len();
    let mut out: Vec<(String, String)> = Vec::new();
    let mut i = 0;
    // `i < n` reads as survivable (`<=` passes the sweep) and is: at
    // `i == n` the skip-whitespace loop below does nothing and the
    // `if i >= n { break }` fires immediately, so the extra iteration has
    // no effect. Equivalent, not a coverage gap.
    while i < n {
        // Skip whitespace (space or tab).
        while i < n && (chars[i] == ' ' || chars[i] == '\t') {
            i += 1;
        }
        if i >= n {
            break;
        }
        // Read key: ident chars.
        let key_start = i;
        while i < n && (chars[i].is_alphanumeric() || chars[i] == '_') {
            i += 1;
        }
        if i == key_start || i >= n || chars[i] != '=' {
            // Not a key=value pair; skip to next whitespace.
            while i < n && chars[i] != ' ' && chars[i] != '\t' {
                i += 1;
            }
            continue;
        }
        let key: String = chars[key_start..i].iter().collect();
        i += 1; // consume '='

        // Read value.
        let value: String = if i < n && chars[i] == '"' {
            // Quoted: everything until the next `"`.
            i += 1;
            let val_start = i;
            while i < n && chars[i] != '"' {
                i += 1;
            }
            let v: String = chars[val_start..i].iter().collect();
            // `<` vs `<=` here is equivalent: the only way to reach this
            // with `i == n` is an unterminated quote, and `i = n + 1`
            // exits the outer loop exactly as `i = n` does. The `==` and
            // `>` forms are NOT equivalent — they skip consuming the
            // closing quote, which swallows the rest of the line when the
            // next key butts straight against it. Pinned by
            // `a_quoted_value_can_butt_against_the_next_key`.
            if i < n {
                i += 1;
            } // consume closing "
            v
        } else {
            // Unquoted: extend until the next `key=` boundary or EOL.
            let val_start = i;
            while i < n {
                if chars[i] == ' ' || chars[i] == '\t' {
                    // Lookahead: does the next non-whitespace chunk
                    // look like `ident=`? If so, stop here.
                    // `i + 1` vs `i * 1`: chars[i] is the whitespace just
                    // matched and the loop below skips it either way, so
                    // both land on the same character. Equivalent.
                    let mut j = i + 1;
                    // Redundant, and deliberately kept. `cargo mutants`
                    // reports this loop's `||` as survivable, and it is:
                    // the outer scan advances one character at a time, so
                    // it reaches the position immediately before the key
                    // anyway, and `.trim()` below erases the extra
                    // whitespace either way. Verified by deleting the loop
                    // outright — all 1222 tests still pass. The
                    // 2026-08-26 sweep reports this loop's `<`, its `||`
                    // and its `+=` as survivable for the same reason —
                    // `.trim()` below absorbs the difference in every
                    // case. All equivalent.
                    //
                    // Not deleted, because this parser is what `ebman
                    // audit replay` reconstructs an AWS action from, and
                    // three redundant lines are a better trade than the
                    // chance that the reasoning above misses an input the
                    // suite does not cover. Treat the mutant as
                    // equivalent rather than as a coverage gap.
                    while j < n && (chars[j] == ' ' || chars[j] == '\t') {
                        j += 1;
                    }
                    let ident_start = j;
                    while j < n && (chars[j].is_alphanumeric() || chars[j] == '_') {
                        j += 1;
                    }
                    if j > ident_start && j < n && chars[j] == '=' {
                        break;
                    }
                }
                i += 1;
            }
            let raw: String = chars[val_start..i].iter().collect();
            raw.trim().to_string()
        };
        out.push((key, value));
    }
    out
}

/// Sanitize a value-string for embedding in an audit-log line as the
/// inner content of a quoted `key="..."` pair. Replaces:
///
/// - `"` → `'` so the closing quote is unambiguous;
/// - `\n`, `\r`, `\t` → ` ` so a multi-line AWS error doesn't split
///   one audit entry into two on disk (the parser reads line-by-line,
///   so an embedded newline corrupts the next entry's RFC3339 prefix).
///
/// Used by `append_action_completed`, `append_rollout`, and
/// `append_lint_fix` (and the typed wrappers in `app.rs` that
/// route to them) so the escape rules stay consistent across every
/// writer.
pub(crate) fn escape_value(s: &str) -> String {
    s.chars()
        .map(|c| match c {
            '"' => '\'',
            '\n' | '\r' | '\t' => ' ',
            c => c,
        })
        .collect()
}

/// Filter spec applied to a parsed audit log. Returned subset is sorted
/// in the same order as the input.
#[derive(Debug, Default, Clone)]
pub(crate) struct AuditFilter<'a> {
    pub since: Option<chrono::DateTime<chrono::Utc>>,
    pub env: Option<&'a str>,
    pub rule: Option<&'a str>,
    pub action: Option<&'a str>,
}

impl<'a> AuditFilter<'a> {
    pub(crate) fn matches(&self, entry: &AuditEntry) -> bool {
        if let Some(since) = self.since {
            if let Ok(when) = chrono::DateTime::parse_from_rfc3339(&entry.when) {
                if when.with_timezone(&chrono::Utc) < since {
                    return false;
                }
            } else {
                return false;
            }
        }
        if let Some(want) = self.env {
            if entry.target.as_deref() != Some(want) {
                return false;
            }
        }
        if let Some(want) = self.rule {
            if entry.rule_id.as_deref() != Some(want) {
                return false;
            }
        }
        if let Some(want) = self.action {
            if entry.action.as_deref() != Some(want) {
                return false;
            }
        }
        true
    }
}

/// Render audit entries as a pretty text table (TS / REGION / STAGE /
/// ACTION / TARGET / OUTCOME). Empty input yields a one-line `(no
/// entries)` so the operator sees the empty result didn't silently
/// match nothing.
pub(crate) fn render_audit_entries_text(entries: &[AuditEntry]) -> String {
    if entries.is_empty() {
        return "(no audit entries)\n".to_string();
    }
    let mut out = String::new();
    // Column widths sized to content.
    let w_ts = entries
        .iter()
        .map(|e| e.when.len())
        .max()
        .unwrap_or(20)
        .max(20);
    let w_region = entries
        .iter()
        .map(|e| e.region.as_deref().unwrap_or("-").len())
        .max()
        .unwrap_or(6)
        .max(6);
    let w_stage = entries
        .iter()
        .map(|e| e.stage.as_deref().unwrap_or("-").len())
        .max()
        .unwrap_or(5)
        .max(5);
    let w_action = entries
        .iter()
        .map(|e| e.action.as_deref().unwrap_or("-").len())
        .max()
        .unwrap_or(6)
        .max(6);
    let w_target = entries
        .iter()
        .map(|e| e.target.as_deref().unwrap_or("-").len())
        .max()
        .unwrap_or(6)
        .max(6);

    out.push_str(&format!(
        "{:<w_ts$}  {:<w_region$}  {:<w_stage$}  {:<w_action$}  {:<w_target$}  OUTCOME\n",
        "TS", "REGION", "STAGE", "ACTION", "TARGET",
    ));
    for e in entries {
        let outcome = match (e.outcome.as_deref(), e.err.as_deref()) {
            (_, Some(err)) => format!("err=\"{err}\""),
            // No special case for "ok": `(Some(s), _)` renders it
            // identically. One was there, and the sweep reported deleting
            // it as survivable — correctly, since it was pure
            // duplication rather than an untested branch.
            (Some(s), _) => s.into(),
            _ => "-".into(),
        };
        out.push_str(&format!(
            "{:<w_ts$}  {:<w_region$}  {:<w_stage$}  {:<w_action$}  {:<w_target$}  {outcome}\n",
            e.when,
            e.region.as_deref().unwrap_or("-"),
            e.stage.as_deref().unwrap_or("-"),
            e.action.as_deref().unwrap_or("-"),
            e.target.as_deref().unwrap_or("-"),
        ));
    }
    out
}

/// Render audit entries as JSON Lines (one JSON object per line).
///
/// Hand-rolled. The original reason — "so we don't pull in `serde_json`
/// for this one path" — stopped being true when five JSON surfaces moved
/// off the YAML parser onto `serde_json`, which is a direct dependency
/// now. It stays hand-rolled for a different reason: the key order here
/// is the declaration order below and consumers grep it, whereas a
/// derived serialisation would tie that order to the struct. Values are
/// escaped by [`crate::util::json_string`], the canonical escaper.
pub(crate) fn render_audit_entries_json(entries: &[AuditEntry]) -> String {
    let mut out = String::new();
    for e in entries {
        let mut first = true;
        out.push('{');
        let mut emit = |key: &str, val: Option<&str>| {
            if let Some(v) = val {
                if !first {
                    out.push(',');
                }
                first = false;
                out.push_str(&format!("\"{key}\":{}", json_string(v)));
            }
        };
        emit("when", Some(&e.when));
        emit("account", e.account.as_deref());
        emit("profile", e.profile.as_deref());
        emit("region", e.region.as_deref());
        emit("rollout_id", e.rollout_id.as_deref());
        emit("stage", e.stage.as_deref());
        emit("action", e.action.as_deref());
        emit("target", e.target.as_deref());
        emit("version", e.version.as_deref());
        emit("rule_id", e.rule_id.as_deref());
        emit("outcome", e.outcome.as_deref());
        emit("err", e.err.as_deref());
        if !e.extras.is_empty() {
            if !first {
                out.push(',');
            }
            out.push_str("\"extras\":{");
            let mut first_extra = true;
            for (k, v) in &e.extras {
                if !first_extra {
                    out.push(',');
                }
                first_extra = false;
                out.push_str(&format!("{}:{}", json_string(k), json_string(v)));
            }
            out.push('}');
        }
        out.push_str("}\n");
    }
    out
}

/// Audit JSONL output uses the canonical [`crate::util::json_string`]
/// for value escaping.
use crate::util::json_string;

// ─── writers ─────────────────────────────────────────────────

/// Soft cap on `audit.log` size before we rotate to `audit.log.1`
/// (single historical backup, older history is discarded). 1 MiB ≈
/// ~5k action entries, plenty for an interactive operator tool.
const AUDIT_LOG_MAX_BYTES: u64 = 1 << 20;

/// Process-wide webhook URL for audit-line fan-out. Set once at App
/// or CLI startup from the resolved Config. `None` (or absent) means
/// no fan-out; the local audit file is always the source of truth.
static NOTIFY_WEBHOOK_URL: std::sync::OnceLock<Option<String>> = std::sync::OnceLock::new();

/// Configure the outbound webhook URL exactly once per process. Idempotent:
/// subsequent calls are no-ops (the first call wins, matching the previous
/// behaviour when this was an `OnceLock::set` site in `App::new`).
pub(crate) fn set_notify_webhook(url: Option<String>) {
    let _ = NOTIFY_WEBHOOK_URL.set(url);
}

/// Load `notify_webhook` from `~/.config/ebman/config.toml` and register
/// it for fan-out. Called once at CLI startup so audit lines emitted
/// by `ebman lint --fix`, `ebman action rollout`, etc. fan out to the
/// same webhook the TUI uses. Idempotent (the OnceLock guards repeat
/// calls). No-op when config can't be read; webhook is optional.
pub fn init_from_config_disk() {
    let cfg = crate::config::load();
    set_notify_webhook(cfg.notify_webhook);
}

/// Append a `stage=dispatched` line for a TUI-driven action. `target`
/// is pre-formatted by the caller so swap (`env-a ↔ env-b`) and
/// single-env (`env-a`) shapes are the caller's choice. `action_label`
/// goes into the `action=` field verbatim — typically the Debug-derived
/// variant name of [`crate::mode_action::Action`].
///
/// `extras` is an optional slice of `(key, value)` pairs emitted
/// after `target=`. Values are auto-quoted with `"..."` when they
/// contain whitespace; [`escape_value`] sanitises them so a stray
/// newline can't split the line on disk. Use this for additional
/// context the simple `action/target` shape doesn't carry (e.g.
/// `version=build-900`, `summary="MinSize=2 MaxSize=4"`).
pub(crate) fn append_action_dispatched(
    account: Option<&str>,
    profile: Option<&str>,
    region: &str,
    action_label: &str,
    target: &str,
    extras: &[(&str, &str)],
) {
    let mut detail = format!(
        "stage=dispatched action={action_label} {}",
        field_token("target", target)
    );
    append_extras(&mut detail, extras);
    write_audit_line(account, profile, region, &detail);
}

/// Append a `stage=completed` line for a TUI-driven action. `result`
/// is mapped to `outcome=ok` (Ok) or `outcome=err err="…"` (Err); the
/// error string goes through [`escape_value`] so a multi-line AWS
/// error doesn't split the entry across two log lines.
///
/// `extras` (same shape as in `append_action_dispatched`) lets
/// callers attach per-action context — e.g. `summary="..."` for an
/// option-settings update, `label=...` for a deploy, `cmd="..."`
/// for an SSM RunCommand. Emitted between `target=` and `outcome=`
/// so the wire shape stays stable.
pub(crate) fn append_action_completed(
    account: Option<&str>,
    profile: Option<&str>,
    region: &str,
    action_label: &str,
    target: &str,
    result: Result<(), &str>,
    extras: &[(&str, &str)],
) {
    let mut detail = format!(
        "stage=completed action={action_label} {}",
        field_token("target", target)
    );
    append_extras(&mut detail, extras);
    match result {
        Ok(()) => detail.push_str(" outcome=ok"),
        Err(e) => detail.push_str(&format!(" outcome=err err=\"{}\"", escape_value(e))),
    }
    write_audit_line(account, profile, region, &detail);
}

/// Append `extras` to a detail string. Pure helper so the
/// dispatched + completed paths share the encoding (and so the
/// tests cover it once). Auto-quotes values that contain
/// whitespace, `=`, or `"`; leaves simple values unquoted to match
/// the existing hand-rolled audit-line shape
/// (`namespace=ns name=opt value="..."`).
/// Render one `key=value` token with the same auto-quoting
/// `append_extras` applies: quote + escape when the value is empty or
/// contains whitespace / `"` / `=` / newline. Free-text fields
/// (target env names, version labels) previously interpolated raw —
/// today's inputs are AWS-constrained so no forge path existed, but a
/// future caller passing free text would have split lines / forged
/// fields (parse_audit_line treats an embedded newline as a new,
/// replayable entry).
fn field_token(key: &str, value: &str) -> String {
    if value.is_empty() || value.contains(|c: char| c.is_whitespace() || c == '"' || c == '=') {
        format!("{key}=\"{}\"", escape_value(value))
    } else {
        format!("{key}={value}")
    }
}

fn append_extras(detail: &mut String, extras: &[(&str, &str)]) {
    for (k, v) in extras {
        if v.is_empty() || v.contains(|c: char| c.is_whitespace() || c == '"' || c == '=') {
            detail.push_str(&format!(" {k}=\"{}\"", escape_value(v)));
        } else {
            detail.push_str(&format!(" {k}={v}"));
        }
    }
}

/// Pure: the tail of a rollout audit line. Extracted so the field set
/// can be asserted without writing to the log — `append_rollout` is the
/// I/O wrapper around it.
fn rollout_line(
    rollout_id: &str,
    profile: Option<&str>,
    region: &str,
    env: &str,
    version: &str,
    stage: &str,
    err: Option<&str>,
) -> String {
    let outcome_suffix = match (stage, err) {
        ("completed", None) => " outcome=ok".to_string(),
        ("completed", Some(e)) => format!(" outcome=err err=\"{}\"", escape_value(e)),
        (_, Some(e)) => format!(" err=\"{}\"", escape_value(e)),
        (_, None) => String::new(),
    };
    // `profile` rides along even though this shape does not use the
    // standard `account=/profile=/region=` opener.
    //
    // Rollout is the only CLI command that takes `--profile`, it is
    // multi-region by construction, and the file that dispatches it
    // calls it "the biggest write the CLI has". So it was the one
    // command whose audit lines did not record which account the write
    // landed in — precisely the question you ask the audit log after an
    // incident. Additive and safe: the parser is key-value and keeps
    // unrecognised keys in `extras`.
    // `field_token`, not bare `escape_value`. `escape_value` maps quotes
    // and newlines but leaves SPACES alone, and these fields are
    // space/tab separated — so a profile literally named
    // `ops region=us-fake-1` would emit a second `region=` token, and
    // every consumer takes the first match. `field_token` quotes
    // anything containing whitespace, `=` or `"`, which `parse_kv_pairs`
    // then reads back as one value.
    //
    // Self-inflicted (the operator names their own profiles) and
    // pre-existing in the header opener, which is fixed the same way
    // below — adding a new field to a shape with this flaw means fixing
    // the class rather than shipping the precedent.
    format!(
        "\t{}\t{}\t{}\tstage={stage} action=Rollout {} {}{outcome_suffix}",
        field_token("rollout_id", rollout_id),
        field_token("profile", profile.unwrap_or("-")),
        field_token("region", region),
        field_token("target", env),
        field_token("version", version)
    )
}

/// Append a rollout-shaped line. `stage` is `"dispatched"` or
/// `"completed"`; pass `err = Some(...)` to attach an error message
/// (and emit `outcome=err` on completion). `rollout_id` correlates
/// every per-region line within a single `ebman action rollout`
/// invocation.
pub(crate) fn append_rollout(
    rollout_id: &str,
    profile: Option<&str>,
    region: &str,
    env: &str,
    version: &str,
    stage: &str,
    err: Option<&str>,
) {
    write_audit_line_raw(&rollout_line(
        rollout_id, profile, region, env, version, stage, err,
    ));
}

/// Append a `stage=fix action=SetOption` line for an `ebman lint
/// --fix` dispatch. `rule_id` correlates back to which lint rule
/// triggered the change so `ebman audit --rule EBL001` shows per-
/// rule history.
pub(crate) fn append_lint_fix(
    region: &str,
    env: &str,
    rule_id: &str,
    namespace: &str,
    name: &str,
    value: &str,
    err: Option<&str>,
) {
    let q_value = escape_value(value);
    let suffix = match err {
        None => " outcome=ok".to_string(),
        Some(e) => format!(" outcome=err err=\"{}\"", escape_value(e)),
    };
    // Every free-text field through `field_token`, like the other two
    // writers. These were the last raw interpolations: `parse_audit_line`
    // treats an embedded newline as a new, REPLAYABLE entry, so a value
    // that could carry one is a forge path into `ebman audit replay`.
    // Today's inputs are AWS-constrained, which is why nothing has gone
    // wrong — but "currently impossible by accident" is not the same
    // property as "escaped".
    let line = format!(
        "\tregion={}\tstage=fix action=SetOption {} {} {} {} value=\"{q_value}\"{suffix}",
        escape_value(region),
        field_token("target", env),
        field_token("rule_id", rule_id),
        field_token("namespace", namespace),
        field_token("name", name),
    );
    write_audit_line_raw(&line);
}

/// Append a `stage=skipped` line — an action that was deliberately not
/// dispatched (e.g. a batch member whose env vanished from the current
/// view mid-run). `reason` is quoted via [`escape_value`]. Same wire
/// shape the batch paths used to hand-roll; lifted here so the format
/// lives in one place alongside the other typed audit helpers.
pub(crate) fn append_action_skipped(
    account: Option<&str>,
    profile: Option<&str>,
    region: &str,
    action_label: &str,
    target: &str,
    reason: &str,
) {
    let detail = format!(
        "stage=skipped action={action_label} {} reason=\"{}\"",
        field_token("target", target),
        escape_value(reason)
    );
    write_audit_line(account, profile, region, &detail);
}

/// Append a `stage=refused` line — a write that policy stopped before
/// it reached AWS.
///
/// Deliberately NOT `stage=skipped`. Skipped means "deliberately not
/// dispatched" for benign operational reasons (a batch member whose env
/// vanished mid-run); refused means a safety control fired. Folding
/// them together would make the log unable to answer the one question
/// it is being extended to answer: did anything try this, and what
/// stopped it. Until now a blocked write left no trace at all — the
/// dispatch never happened, so no dispatched/completed pair was ever
/// written, and six attempts to terminate prod looked exactly like
/// none.
///
/// `rule` is the machine token from `write_gate::Refusal::rule`;
/// `remedy` says which control would have to change.
pub(crate) fn append_action_refused(
    account: Option<&str>,
    profile: Option<&str>,
    region: &str,
    action_label: &str,
    target: &str,
    rule: &str,
    remedy: &str,
) {
    // `action` goes through `field_token` like `target`, not raw. It
    // carries operator-chosen text on some paths (a rename summary, a
    // batch verb), and `escape_value` does not quote spaces — so a bare
    // interpolation lets a crafted value forge a `stage=` token that
    // every consumer reads instead of the real one. Same forge path
    // `field_token`'s own comment exists to close for the header
    // fields.
    let detail = format!(
        "stage=refused {} {} rule={rule} remedy=\"{}\"",
        field_token("action", action_label),
        field_token("target", target),
        escape_value(remedy)
    );
    write_audit_line(account, profile, region, &detail);
}

/// Append a `stage=undone` line — an operator-driven undo of a prior
/// action. No outcome (the undo dispatch logs its own completion via the
/// normal action path); this records that the undo was initiated.
pub(crate) fn append_action_undone(
    account: Option<&str>,
    profile: Option<&str>,
    region: &str,
    action_label: &str,
    target: &str,
) {
    let detail = format!(
        "stage=undone action={action_label} {}",
        field_token("target", target)
    );
    write_audit_line(account, profile, region, &detail);
}

/// Append a one-shot DLQ operation line (delete / resend / purge /
/// replay). These have no dispatched+completed pair — they're recorded
/// at fire time. `op` is the verb (`sqs-delete` / `dlq-resend` /
/// `dlq-purge` / `dlq-replay`); `extras` carry per-op context
/// (`msg_id`, `queue`, `count`) and are encoded like every other typed
/// helper. Centralizes the format the four DLQ spawn sites hand-rolled.
pub(crate) fn append_dlq_op(
    account: Option<&str>,
    profile: Option<&str>,
    region: &str,
    op: &str,
    env: &str,
    extras: &[(&str, &str)],
) {
    write_audit_line(account, profile, region, &dlq_op_detail(op, env, extras));
}

/// Pure detail-builder behind `append_dlq_op` — `"{op} env={env}"`
/// plus encoded extras. Split out so the wire shape is unit-testable
/// without the file-write side effect.
fn dlq_op_detail(op: &str, env: &str, extras: &[(&str, &str)]) -> String {
    let mut detail = format!("{op} env={env}");
    append_extras(&mut detail, extras);
    detail
}

/// Build the JSON body that goes to `notify_webhook`. Pure +
/// deterministic so the shape is unit-testable. Top-level `text`
/// gets the rendered audit line so the body is
/// Slack-incoming-webhook-compatible out of the box; the other
/// keys give consumers structured fields for routing / filtering.
pub(crate) fn build_webhook_body(
    account: Option<&str>,
    profile: Option<&str>,
    region: &str,
    detail: &str,
    when: &str,
) -> String {
    let text = format!(
        "[ebman] {} account={} profile={} region={} {}",
        when,
        account.unwrap_or("-"),
        profile.unwrap_or("-"),
        region,
        detail,
    );
    format!(
        "{{\"text\":\"{}\",\"at\":\"{}\",\"account\":\"{}\",\"profile\":\"{}\",\"region\":\"{}\",\"detail\":\"{}\"}}",
        json_escape(&text),
        json_escape(when),
        json_escape(account.unwrap_or("")),
        json_escape(profile.unwrap_or("")),
        json_escape(region),
        json_escape(detail),
    )
}

/// Append a raw audit-log line with a caller-built `detail` string.
/// Used by sites that emit non-action lines (red-transition events,
/// notifications, etc.) where the typed `append_action_*` APIs
/// don't fit. The `detail` string is appended verbatim after the
/// `account/profile/region` opener — caller is responsible for the
/// `key=value` shape + escaping.
pub(crate) fn append_raw(account: Option<&str>, profile: Option<&str>, region: &str, detail: &str) {
    write_audit_line(account, profile, region, detail);
}

fn write_audit_line(account: Option<&str>, profile: Option<&str>, region: &str, detail: &str) {
    let dir = crate::util::cache_dir();
    if std::fs::create_dir_all(&dir).is_err() {
        return;
    }
    let path = dir.join("audit.log");
    rotate_if_oversize(&path, AUDIT_LOG_MAX_BYTES);
    let when = chrono::Utc::now().to_rfc3339();
    // The header fields go through `escape_value` too. `account` and
    // `region` are AWS-constrained, but `profile` is whatever the
    // operator named a section in `~/.aws/config` — free text, and
    // `\t` is the field separator this format is parsed on.
    // See `rollout_line`: `escape_value` does not quote spaces, and a
    // profile name is operator-chosen free text, so a bare interpolation
    // here can forge a `region=` token that every consumer reads instead
    // of the real one.
    let line = format!(
        "{when}\t{}\t{}\t{}\t{detail}\n",
        field_token("account", account.unwrap_or("-")),
        field_token("profile", profile.unwrap_or("-")),
        field_token("region", region),
    );
    use std::io::Write;
    if let Ok(mut f) = crate::util::open_append_secure(&path) {
        let _ = f.write_all(line.as_bytes());
    }
    // Webhook fan-out — same convention as before consolidation.
    if let Some(url) = NOTIFY_WEBHOOK_URL.get().and_then(|o| o.as_deref()) {
        fire_webhook(url, account, profile, region, detail, &when);
    }
}

/// Lower-level append: caller has already constructed the tab-prefixed
/// `\tkey=value\t...\tstage=... ...` tail (no leading timestamp). Used
/// by line shapes that don't follow the standard
/// `account=A\tprofile=P\tregion=R` opener (rollout uses
/// `rollout_id=...\tregion=...`; lint-fix uses just `region=...`).
fn write_audit_line_raw(tail: &str) {
    let dir = crate::util::cache_dir();
    if std::fs::create_dir_all(&dir).is_err() {
        return;
    }
    let path = dir.join("audit.log");
    rotate_if_oversize(&path, AUDIT_LOG_MAX_BYTES);
    let when = chrono::Utc::now().to_rfc3339();
    let line = format!("{when}{tail}\n");
    use std::io::Write;
    if let Ok(mut f) = crate::util::open_append_secure(&path) {
        let _ = f.write_all(line.as_bytes());
    }
    // Same webhook fan-out as `write_audit_line`. The body uses
    // `account=-, profile=-` because this shape doesn't carry them;
    // consumers route on `detail` / `region` instead.
    if let Some(url) = NOTIFY_WEBHOOK_URL.get().and_then(|o| o.as_deref()) {
        // Strip the leading tab so the detail string is the same
        // shape webhook consumers expect (key=value space-separated).
        let detail = tail.trim_start_matches('\t').replace('\t', " ");
        // Same extraction for `profile` as for `region`, and for the
        // same reason. Rollout lines gained a `profile=` field but this
        // still passed `None`, so a rollout webhook contradicted itself:
        // the structured `profile` was empty while `detail` right beside
        // it read `profile=prod`. The `region` line above is the
        // precedent — a field added to the line shape has to be chased
        // here too, or the webhook and the log disagree.
        let region = detail_field(&detail, "region=").unwrap_or("-");
        let profile = detail_field(&detail, "profile=");
        fire_webhook(url, None, profile, region, &detail, &when);
    }
}

/// Pull `key=value` out of an already-rendered audit `detail` string,
/// treating an empty value and the placeholder `-` as absent.
///
/// Was a closure inside `write_audit_line_raw`, which meant the only way
/// to reach it was to fire a webhook. It is the piece that decides what
/// the webhook reports, and it has already been wrong once: the
/// structured `profile` went out empty while `detail` right beside it
/// read `profile=prod`. Worth being able to test on its own.
fn detail_field<'a>(detail: &'a str, key: &str) -> Option<&'a str> {
    detail
        .split(' ')
        .find_map(|tok| tok.strip_prefix(key))
        .filter(|v| !v.is_empty() && *v != "-")
}

/// Fire-and-forget webhook POST via `reqwest` (the same HTTP client
/// `llm.rs` already pulls in, so no extra dependency). 10s timeout so
/// a slow webhook can't accumulate hung requests. The caller must be
/// inside a tokio runtime — guarded below with `Handle::try_current`
/// so a non-runtime call path silently no-ops rather than panicking.
/// `pub(crate)` so `ebman lint --watch --webhook URL` can post its
/// per-cycle findings through the same body shape + client settings
/// as the audit-line fan-out.
pub(crate) fn fire_webhook(
    url: &str,
    account: Option<&str>,
    profile: Option<&str>,
    region: &str,
    detail: &str,
    when: &str,
) {
    let body = build_webhook_body(account, profile, region, detail, when);
    let url = url.to_string();
    if tokio::runtime::Handle::try_current().is_err() {
        return;
    }
    WEBHOOKS_IN_FLIGHT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
    tokio::spawn(async move {
        // Decrement-on-drop so every exit path (success, error, panic)
        // releases the drain counter.
        struct InFlight;
        impl Drop for InFlight {
            fn drop(&mut self) {
                WEBHOOKS_IN_FLIGHT.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
            }
        }
        let _guard = InFlight;
        // In the TUI, failures go to tracing only (stderr would tear
        // the alternate screen). CLI subcommands may print — the ones
        // that take an operator-supplied webhook opt in via
        // `webhook_errors_to_stderr()` so a broken URL isn't silently
        // swallowed forever (CLI runs install no tracing subscriber).
        let to_stderr = WEBHOOK_ERRORS_TO_STDERR.load(std::sync::atomic::Ordering::Relaxed);
        let client = match reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(10))
            .build()
        {
            Ok(c) => c,
            Err(e) => {
                tracing::warn!(
                    target: "ebman::notify",
                    url = %url,
                    error = %e,
                    "audit webhook: could not build reqwest client"
                );
                if to_stderr {
                    eprintln!("warning: webhook client build failed: {e}");
                }
                return;
            }
        };
        match client
            .post(&url)
            .header("Content-Type", "application/json")
            .body(body)
            .send()
            .await
        {
            Ok(resp) if resp.status().is_success() => {}
            Ok(resp) => {
                tracing::warn!(
                    target: "ebman::notify",
                    url = %url,
                    status = %resp.status(),
                    "audit webhook returned non-success status"
                );
                if to_stderr {
                    eprintln!("warning: webhook POST returned {}", resp.status());
                }
            }
            Err(e) => {
                tracing::warn!(
                    target: "ebman::notify",
                    url = %url,
                    error = %e,
                    "audit webhook request failed"
                );
                if to_stderr {
                    eprintln!("warning: webhook POST failed: {e}");
                }
            }
        }
    });
}

/// See the comment in [`fire_webhook`]: CLI paths that take an
/// operator-supplied webhook URL flip this so delivery failures reach
/// stderr. Never set from TUI code.
static WEBHOOK_ERRORS_TO_STDERR: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

pub(crate) fn webhook_errors_to_stderr() {
    WEBHOOK_ERRORS_TO_STDERR.store(true, std::sync::atomic::Ordering::Relaxed);
}

/// Count of webhook POSTs currently in flight. `fire_webhook` is
/// fire-and-forget, which is right for the TUI — but one-shot CLI
/// commands return from `#[tokio::main]` immediately after their last
/// audit line, and runtime drop CANCELS spawned tasks: the outcome
/// POST (the line a paging integration most needs) usually never left
/// the machine. CLI exits call [`drain_webhooks`] first.
static WEBHOOKS_IN_FLIGHT: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);

/// Serialises the tests that drive `WEBHOOKS_IN_FLIGHT`.
///
/// The counter is process-global, so two tests setting it at once would
/// see each other's value — the same shape as the audit-log race a
/// 2026-08-26 review found, where a test asserted against a neighbour's
/// data. Only `fire_webhook` touches it in production, and no test
/// reaches that (it makes a real request), so this exists purely to keep
/// the tests below from racing each other as more are added.
/// A tokio mutex, not a `std` one: these tests await inside the guard,
/// and clippy rightly refuses a `std::sync::MutexGuard` held across an
/// await point. `aws::CACHE_TEST_LOCK` is the same shape for the same
/// reason.
#[cfg(test)]
pub(crate) static WEBHOOK_COUNTER_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

/// Wait (bounded) for in-flight webhook POSTs to finish. Call before
/// a one-shot CLI command returns/exits; a no-op when no webhook is
/// configured or nothing is in flight. Polling is fine here: the
/// steady state is "zero in flight", and the worst case is one
/// 10s-timeout POST.
pub async fn drain_webhooks(max_wait: std::time::Duration) {
    let deadline = tokio::time::Instant::now() + max_wait;
    while WEBHOOKS_IN_FLIGHT.load(std::sync::atomic::Ordering::SeqCst) > 0 {
        if tokio::time::Instant::now() >= deadline {
            tracing::warn!(target: "ebman::notify", "webhook drain timed out with POSTs in flight");
            return;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
}

/// If `path` exists and is larger than `max_bytes`, move it to
/// `path.1` (overwriting any previous backup) so the next write
/// starts a fresh file. Best-effort: any I/O error is swallowed —
/// we don't want to lose the audit entry just because rotation
/// failed.
fn rotate_if_oversize(path: &std::path::Path, max_bytes: u64) {
    let Ok(meta) = std::fs::metadata(path) else {
        return;
    };
    if meta.len() <= max_bytes {
        return;
    }
    let backup = {
        let mut name = path
            .file_name()
            .map(|s| s.to_os_string())
            .unwrap_or_default();
        name.push(".1");
        path.with_file_name(name)
    };
    let _ = std::fs::rename(path, backup);
}

// Webhook body uses the canonical `crate::util::json_escape`; the
// previously-local `json_escape` is routed through there via the
// import at the top of the writers section.
use crate::util::json_escape;

#[cfg(test)]
mod tests {
    #[test]
    fn free_text_fields_cannot_split_or_forge_lines() {
        // A newline-bearing target must stay ONE parseable entry —
        // an embedded newline used to become a second (replayable)
        // audit line.
        let mut detail = format!(
            "stage=dispatched action=Deploy {}",
            super::field_token(
                "target",
                "evil\nstage=completed action=Terminate target=prod"
            )
        );
        super::append_extras(&mut detail, &[]);
        assert!(!detail.contains('\n'), "newline must be escaped: {detail}");
        let line = format!("2026-08-20T00:00:00+00:00\taccount=1\tprofile=-\tregion=r\t{detail}");
        let entry = super::parse_audit_line(&line).expect("one entry");
        assert_eq!(entry.action.as_deref(), Some("Deploy"));
        assert!(entry
            .target
            .as_deref()
            .unwrap_or_default()
            .starts_with("evil"));
    }

    use super::*;

    #[test]
    fn dlq_op_detail_preserves_wire_shape() {
        // Format must match what the four DLQ spawn sites hand-rolled
        // before centralizing — `ebman audit` parses these lines.
        assert_eq!(
            dlq_op_detail("dlq-replay", "prod", &[("count", "12")]),
            "dlq-replay env=prod count=12"
        );
        assert_eq!(
            dlq_op_detail("dlq-purge", "prod", &[]),
            "dlq-purge env=prod"
        );
        assert_eq!(
            dlq_op_detail("sqs-delete", "prod", &[("queue", "DLQ"), ("msg_id", "abc")]),
            "sqs-delete env=prod queue=DLQ msg_id=abc"
        );
    }

    #[test]
    fn parse_kv_simple_pairs() {
        let pairs = parse_kv_pairs("account=A profile=B region=us-east-1");
        assert_eq!(
            pairs,
            vec![
                ("account".into(), "A".into()),
                ("profile".into(), "B".into()),
                ("region".into(), "us-east-1".into()),
            ]
        );
    }

    #[test]
    fn parse_kv_quoted_value() {
        let pairs = parse_kv_pairs("err=\"AccessDenied: not authorized\" stage=completed");
        assert_eq!(
            pairs,
            vec![
                ("err".into(), "AccessDenied: not authorized".into()),
                ("stage".into(), "completed".into()),
            ]
        );
    }

    #[test]
    fn parse_kv_unquoted_value_with_spaces() {
        // The naked space before "↔" should be preserved because the
        // next token isn't `key=`.
        let pairs = parse_kv_pairs("target=env-a ↔ env-b stage=dispatched");
        assert_eq!(
            pairs,
            vec![
                ("target".into(), "env-a ↔ env-b".into()),
                ("stage".into(), "dispatched".into()),
            ]
        );
    }

    #[test]
    fn parse_kv_tab_separator() {
        let pairs = parse_kv_pairs("account=A\tprofile=B\tregion=R");
        assert_eq!(
            pairs,
            vec![
                ("account".into(), "A".into()),
                ("profile".into(), "B".into()),
                ("region".into(), "R".into()),
            ]
        );
    }

    #[test]
    fn parse_audit_line_normal_dispatched() {
        let line = "2026-05-27T10:15:30Z\taccount=123\tprofile=prod\tregion=us-east-1\tstage=dispatched action=Restart target=my-env";
        let entry = crate::audit::parse_audit_line(line).expect("parses");
        assert_eq!(entry.when, "2026-05-27T10:15:30Z");
        assert_eq!(entry.account.as_deref(), Some("123"));
        assert_eq!(entry.profile.as_deref(), Some("prod"));
        assert_eq!(entry.region.as_deref(), Some("us-east-1"));
        assert_eq!(entry.stage.as_deref(), Some("dispatched"));
        assert_eq!(entry.action.as_deref(), Some("Restart"));
        assert_eq!(entry.target.as_deref(), Some("my-env"));
        assert!(entry.err.is_none());
        assert!(entry.outcome.is_none());
    }

    #[test]
    fn parse_audit_line_completed_with_outcome_ok() {
        // Modern shape: `outcome=ok` as an explicit key=value pair.
        // 0.14+ writers emit this so the parser doesn't have to
        // special-case bare trailing "ok".
        let line = "2026-05-27T10:15:31Z\taccount=123\tprofile=prod\tregion=us-east-1\tstage=completed action=Restart target=my-env outcome=ok";
        let entry = crate::audit::parse_audit_line(line).expect("parses");
        assert_eq!(entry.stage.as_deref(), Some("completed"));
        assert_eq!(entry.action.as_deref(), Some("Restart"));
        assert_eq!(entry.target.as_deref(), Some("my-env"));
        assert_eq!(entry.outcome.as_deref(), Some("ok"));
    }

    #[test]
    fn parse_audit_line_pre_0_14_bare_ok_lossy_but_parses() {
        // Pre-0.14 entries had bare `ok` after the detail. Parser
        // can't promote it; target value extends to include it. We
        // accept this as a soft regression on legacy log lines —
        // operators who care about historical analysis read the
        // `raw` field.
        let line = "2026-05-26T08:00:00Z\taccount=123\tprofile=prod\tregion=us-east-1\tstage=completed action=Restart target=my-env ok";
        let entry = crate::audit::parse_audit_line(line).expect("parses");
        assert_eq!(entry.outcome, None);
        assert_eq!(entry.target.as_deref(), Some("my-env ok"));
    }

    #[test]
    fn parse_audit_line_completed_with_outcome_err() {
        let line = "2026-05-27T10:16:00Z\taccount=123\tprofile=-\tregion=us-east-1\tstage=completed action=Deploy target=my-env err=\"UpdateEnvironment: throttled\"";
        let entry = crate::audit::parse_audit_line(line).expect("parses");
        assert_eq!(entry.profile, None); // "-" promoted to None
        assert_eq!(entry.err.as_deref(), Some("UpdateEnvironment: throttled"));
    }

    #[test]
    fn parse_audit_line_rollout_shape() {
        let line = "2026-05-27T10:20:00Z\trollout_id=rollout-20260527T102000Z\tregion=eu-west-1\tstage=dispatched action=Rollout target=prod-api version=build-900";
        let entry = crate::audit::parse_audit_line(line).expect("parses");
        assert_eq!(
            entry.rollout_id.as_deref(),
            Some("rollout-20260527T102000Z")
        );
        assert_eq!(entry.region.as_deref(), Some("eu-west-1"));
        assert_eq!(entry.action.as_deref(), Some("Rollout"));
        assert_eq!(entry.target.as_deref(), Some("prod-api"));
        assert_eq!(entry.version.as_deref(), Some("build-900"));
        // No account / profile in rollout shape — should stay None.
        assert!(entry.account.is_none());
        assert!(entry.profile.is_none());
    }

    #[test]
    fn parse_audit_line_blank_returns_none() {
        assert!(parse_audit_line("").is_none());
        assert!(parse_audit_line("   ").is_none());
        assert!(parse_audit_line("\n").is_none());
    }

    #[test]
    fn parse_audit_line_missing_timestamp_returns_none() {
        assert!(parse_audit_line("garbage line without rfc3339").is_none());
    }

    #[test]
    fn filter_by_since() {
        let entries = [
            parse_audit_line("2026-05-27T08:00:00Z\tregion=r\tstage=s action=A target=t").unwrap(),
            parse_audit_line("2026-05-27T11:00:00Z\tregion=r\tstage=s action=A target=t").unwrap(),
        ];
        let cutoff = chrono::DateTime::parse_from_rfc3339("2026-05-27T10:00:00Z")
            .unwrap()
            .with_timezone(&chrono::Utc);
        let filter = AuditFilter {
            since: Some(cutoff),
            ..Default::default()
        };
        let kept: Vec<_> = entries.iter().filter(|e| filter.matches(e)).collect();
        assert_eq!(kept.len(), 1);
        assert_eq!(kept[0].when, "2026-05-27T11:00:00Z");
    }

    #[test]
    fn filter_by_env_target() {
        let entries = [
            parse_audit_line("2026-05-27T10:00:00Z\tregion=r\tstage=s action=Restart target=env-a")
                .unwrap(),
            parse_audit_line("2026-05-27T10:01:00Z\tregion=r\tstage=s action=Restart target=env-b")
                .unwrap(),
        ];
        let filter = AuditFilter {
            env: Some("env-b"),
            ..Default::default()
        };
        let kept: Vec<_> = entries.iter().filter(|e| filter.matches(e)).collect();
        assert_eq!(kept.len(), 1);
        assert_eq!(kept[0].target.as_deref(), Some("env-b"));
    }

    #[test]
    fn filter_by_rule_id() {
        let entries = [
            parse_audit_line("2026-05-27T10:00:00Z\tregion=r\tstage=fix action=SetOption target=env-a rule_id=EBL001")
                .unwrap(),
            parse_audit_line("2026-05-27T10:01:00Z\tregion=r\tstage=fix action=SetOption target=env-a rule_id=EBL004")
                .unwrap(),
        ];
        let filter = AuditFilter {
            rule: Some("EBL004"),
            ..Default::default()
        };
        let kept: Vec<_> = entries.iter().filter(|e| filter.matches(e)).collect();
        assert_eq!(kept.len(), 1);
        assert_eq!(kept[0].rule_id.as_deref(), Some("EBL004"));
    }

    #[test]
    fn rotate_if_oversize_renames_when_too_big() {
        let dir = std::env::temp_dir().join(format!("ebman-rotate-{}", std::process::id()));
        let _ = std::fs::create_dir_all(&dir);
        let path = dir.join("audit.log");
        let backup = dir.join("audit.log.1");
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&backup);
        std::fs::write(&path, vec![b'x'; 100]).unwrap();
        rotate_if_oversize(&path, 50);
        assert!(!path.exists(), "current file should have been renamed");
        assert!(backup.exists(), "rotated backup should now exist");
        let _ = std::fs::remove_file(&backup);
        let _ = std::fs::remove_dir(&dir);
    }

    #[test]
    fn rotate_if_oversize_leaves_small_files_alone() {
        let dir = std::env::temp_dir().join(format!("ebman-rotate-small-{}", std::process::id()));
        let _ = std::fs::create_dir_all(&dir);
        let path = dir.join("audit.log");
        let _ = std::fs::remove_file(&path);
        std::fs::write(&path, b"tiny").unwrap();
        rotate_if_oversize(&path, 1_000);
        assert!(path.exists());
        assert!(!dir.join("audit.log.1").exists());
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_dir(&dir);
    }

    #[test]
    fn build_webhook_body_has_slack_compatible_text_plus_structured_fields() {
        // Slack incoming webhooks consume a top-level `text` field;
        // anything else is metadata for other consumers. Both must
        // be present so one endpoint can serve both.
        let body = build_webhook_body(
            Some("123456789012"),
            Some("prod"),
            "us-east-1",
            "stage=request action=Deploy target=prod-api",
            "2026-05-25T12:00:00Z",
        );
        assert!(body.starts_with('{') && body.ends_with('}'));
        assert!(
            body.contains("\"text\":\"[ebman]"),
            "missing slack-shaped text field"
        );
        assert!(body.contains("\"at\":\"2026-05-25T12:00:00Z\""));
        assert!(body.contains("\"account\":\"123456789012\""));
        assert!(body.contains("\"profile\":\"prod\""));
        assert!(body.contains("\"region\":\"us-east-1\""));
        assert!(body.contains("\"detail\":\"stage=request action=Deploy target=prod-api\""));
    }

    #[test]
    fn build_webhook_body_dashes_missing_account_and_profile_in_text() {
        let body = build_webhook_body(
            None,
            None,
            "eu-west-1",
            "stage=event kind=red_transition env=prod-api",
            "2026-05-25T12:00:00Z",
        );
        assert!(
            body.contains("account=- profile=- region=eu-west-1"),
            "missing dash placeholders in text, got: {body}"
        );
        // Structured fields use empty strings, not "-", so consumers
        // can distinguish "unknown" from "literal dash".
        assert!(body.contains("\"account\":\"\""));
        assert!(body.contains("\"profile\":\"\""));
    }

    #[test]
    fn build_webhook_body_escapes_quotes_in_detail() {
        let body = build_webhook_body(
            None,
            None,
            "us-east-1",
            "stage=event message=\"deploy started\"",
            "2026-05-25T12:00:00Z",
        );
        // Escaped string appears in both `text` and `detail`.
        assert!(body.contains("\\\"deploy started\\\""));
        // A JSON parser: the webhook body IS JSON, and a YAML reader
        // accepts things JSON rejects — so this asserted less than it
        // appeared to.
        let _: serde_json::Value = serde_json::from_str(&body)
            .expect("webhook body must be parseable JSON / YAML-superset");
    }

    #[test]
    fn no_audit_writer_can_be_made_to_forge_a_second_line() {
        // `parse_audit_line` treats an embedded newline as a new entry,
        // and `ebman audit replay` re-dispatches parsed entries — so a
        // writer that interpolates free text raw is a forge path into
        // a destructive command. Today's inputs are AWS-constrained,
        // which is why nothing has gone wrong; "impossible by
        // accident" is not the property we want to rely on.
        const FORGE: &str = "ok\nstage=completed action=Terminate target=prod";

        // The two `field_token` writers.
        assert!(
            !super::field_token("target", FORGE).contains('\n'),
            "target"
        );
        assert!(
            !super::field_token("version", FORGE).contains('\n'),
            "version"
        );
        // A value with no whitespace still can't smuggle a quote or an
        // `=` out of the quoted branch.
        assert!(
            !super::field_token("target", "a\"b=c").contains('"')
                || super::field_token("target", "a\"b=c").matches('"').count() == 2
        );

        // Every writer that reaches the log, driven end to end.
        //
        // `audit.log` is process-global, so other tests append to it in
        // parallel — the assertion is therefore a property of every
        // line rather than a count of them. A forged newline shows up
        // as a line that does NOT start with an RFC3339 timestamp,
        // which is exactly what `parse_audit_line` would then read as
        // a separate, replayable entry.
        let path = crate::util::cache_dir().join("audit.log");
        super::append_action_dispatched(Some(FORGE), Some(FORGE), FORGE, "Restart", FORGE, &[]);
        // `profile` is FORGE too: it is a new field on this line shape, and
        // an unescaped field is an injection vector. Passing the forged
        // value here is what makes this guard cover it.
        super::append_rollout(FORGE, Some(FORGE), FORGE, FORGE, FORGE, "dispatched", None);
        super::append_lint_fix(FORGE, FORGE, FORGE, FORGE, FORGE, FORGE, None);

        let body = std::fs::read_to_string(&path).expect("audit log written");
        let mut ours = 0usize;
        for line in body.lines().filter(|l| !l.is_empty()) {
            assert!(
                line.starts_with("20") && line[..24].contains('T'),
                "a line without a timestamp is a forged entry: {line}"
            );
            assert!(
                super::parse_audit_line(line).is_some(),
                "every line stays parseable: {line}"
            );
            if line.contains("Terminate") {
                ours += 1;
                assert!(
                    !line.contains("\tstage=completed action=Terminate"),
                    "a forged field escaped its quoting: {line}"
                );
            }
        }
        assert!(ours >= 3, "the three writers' lines are in there");
    }

    // ── sweep triage, 2026-08-26 ──────────────────────────────────────
    //
    // The first complete whole-tree mutation sweep left 42 survivors in
    // this file. These cover the ones that turned out to be real; the
    // equivalents are reasoned about at the code they mutate rather than
    // listed here, so the next triage doesn't redo the analysis.

    /// A quoted value butted straight against the next key.
    ///
    /// Four survivors lived here, all in "consume the closing quote":
    /// `if i < n` flipped to `==` or `>` (never consume), and the `i += 1`
    /// flipped to `-=` or `*=`. Every existing test put a space after the
    /// closing quote, and with a space the parser recovers either way —
    /// the stray `"` is skipped as a non-key token. Without one, the rest
    /// of the line is swallowed and `ebman audit replay` silently
    /// reconstructs a different action.
    #[test]
    fn a_quoted_value_can_butt_against_the_next_key() {
        assert_eq!(
            parse_kv_pairs(r#"a="1"b=2"#),
            vec![("a".into(), "1".into()), ("b".into(), "2".into())]
        );
        // The empty-value form has the same shape and is what
        // `field_token` emits for an empty field.
        assert_eq!(
            parse_kv_pairs(r#"a=""b=2"#),
            vec![("a".into(), String::new()), ("b".into(), "2".into())]
        );
    }

    /// An unterminated quote must still yield what it has, rather than
    /// dropping the field or running off the end.
    #[test]
    fn an_unterminated_quote_keeps_the_value() {
        assert_eq!(
            parse_kv_pairs(r#"a="unterminated"#),
            vec![("a".into(), "unterminated".into())]
        );
    }

    /// The timestamp sanity check is a length *floor*, and both of its
    /// boundaries survived: `< 10` flipped to `<= 10` (rejects a bare
    /// date, which is a legal RFC3339 prefix) and to `== 10` (accepts
    /// `2026-8`, which is not a timestamp at all but has a `-` in the
    /// right place).
    #[test]
    fn the_timestamp_floor_is_exactly_ten_characters() {
        let ten = parse_audit_line("2026-08-26\tstage=x");
        assert_eq!(
            ten.map(|e| e.when),
            Some("2026-08-26".to_string()),
            "a bare date is exactly 10 chars and must be accepted"
        );
        assert!(
            parse_audit_line("2026-8\tstage=x").is_none(),
            "6 chars with a dash at index 4 is not a timestamp"
        );
    }

    /// `field_token` quotes on whitespace, `"` **or `=`**, and the `=`
    /// half was untested. It is the half that matters: an unquoted value
    /// containing `=` parses back as two fields, which is field forgery
    /// in a log `audit replay` acts on.
    #[test]
    fn field_token_quotes_on_every_trigger() {
        // Each trigger needs a value that trips ONLY that one. The first
        // version of this test used "a=b" for the `=` case and stopped
        // there, and the mutation run showed why that is not enough:
        // flipping the first `||` to `&&` collapses the predicate to
        // "contains `=`" — `c.is_whitespace() && c == \'"\'` can never be
        // true — and a value containing `=` is quoted either way. A case
        // per trigger, with no overlap.
        for (label, value) in [
            ("whitespace", "env-a ↔ env-b"),
            ("embedded quote", r#"say "hi""#),
            ("equals", "a=b"),
            ("empty", ""),
        ] {
            let token = field_token("target", value);
            assert!(
                token.starts_with(r#"target=""#) && token.ends_with('"'),
                "{label}: {value:?} must be quoted, got {token}"
            );
        }

        // A value tripping none of them stays bare — otherwise "always
        // quote" would pass everything above.
        assert_eq!(field_token("target", "api-prod"), "target=api-prod");

        // The property that matters, stated against the parser rather
        // than the spelling: one field in, one field out.
        assert_eq!(
            parse_kv_pairs(&field_token("target", "a=b")),
            vec![("target".into(), "a=b".into())],
            "an unquoted `=` would forge a second field"
        );
        assert_eq!(
            parse_kv_pairs(&field_token("target", "env-a env-b")),
            vec![("target".into(), "env-a env-b".into())],
            "an unquoted space is at best fragile and at worst a second field"
        );
    }

    /// `detail_field` decides what the webhook reports. Empty and `-`
    /// both mean absent — three survivors sat on that filter (`delete !`,
    /// `&&` to `||`, `!=` to `==`), and it has been wrong before: the
    /// structured `profile` went out empty while `detail` beside it read
    /// `profile=prod`.
    #[test]
    fn detail_field_treats_empty_and_dash_as_absent() {
        let d = "region=us-east-1 profile=prod action=Deploy";
        assert_eq!(detail_field(d, "region="), Some("us-east-1"));
        assert_eq!(detail_field(d, "profile="), Some("prod"));
        assert_eq!(detail_field(d, "nothing="), None);

        assert_eq!(
            detail_field("profile=- region=eu-west-1", "profile="),
            None,
            "`-` is the placeholder for absent, not a profile named `-`"
        );
        assert_eq!(
            detail_field("profile= region=eu-west-1", "profile="),
            None,
            "an empty value is absent"
        );
    }

    fn an_entry(outcome: Option<&str>, err: Option<&str>) -> AuditEntry {
        AuditEntry {
            when: "2026-08-26T10:00:00Z".into(),
            account: Some("123456789012".into()),
            profile: None,
            region: Some("us-east-1".into()),
            rollout_id: None,
            stage: Some("completed".into()),
            action: Some("Deploy".into()),
            target: Some("api-prod".into()),
            version: None,
            rule_id: None,
            outcome: outcome.map(Into::into),
            err: err.map(Into::into),
            extras: Default::default(),
            raw: String::new(),
        }
    }

    /// The text renderer's `(Some(s), _)` arm was deletable: every test
    /// used `ok` or an error, and `ok` is also what the deleted arm
    /// produced, so nothing noticed that any other outcome fell through
    /// to `-`.
    #[test]
    fn a_non_ok_outcome_renders_as_itself() {
        let out = render_audit_entries_text(&[an_entry(Some("skipped"), None)]);
        assert!(
            out.contains("skipped"),
            "a `skipped` outcome must not render as `-`:\n{out}"
        );
        // The two neighbouring arms still work.
        assert!(render_audit_entries_text(&[an_entry(Some("ok"), None)]).contains("ok"));
        assert!(
            render_audit_entries_text(&[an_entry(None, Some("AccessDenied"))])
                .contains("AccessDenied")
        );
        assert!(
            render_audit_entries_text(&[an_entry(None, None)]).contains('-'),
            "no outcome and no error renders as the placeholder"
        );
    }

    /// Four `delete !` survivors in the JSON renderer, all comma
    /// placement. Nothing was parsing the output — the tests asserted on
    /// substrings, which survive a stray or missing comma. Parse it.
    #[test]
    fn the_json_render_is_valid_json() {
        let mut with_extras = an_entry(Some("ok"), None);
        with_extras.extras.insert("bundle".into(), "app.zip".into());
        with_extras.extras.insert("size".into(), "12".into());

        // JSON *Lines*: one object per line, not an array.
        let parse_all = |out: &str| -> Vec<serde_json::Value> {
            out.lines()
                .map(|l| {
                    serde_json::from_str(l)
                        .unwrap_or_else(|e| panic!("not valid JSON ({e}): {l:?}"))
                })
                .collect()
        };

        for (label, entries) in [
            ("empty", vec![]),
            ("one", vec![an_entry(Some("ok"), None)]),
            (
                "several",
                vec![an_entry(Some("ok"), None), an_entry(None, Some("boom"))],
            ),
            ("with extras", vec![with_extras.clone()]),
        ] {
            let out = render_audit_entries_json(&entries);
            let objs = parse_all(&out);
            assert_eq!(objs.len(), entries.len(), "{label}: wrong line count");
            for o in &objs {
                assert!(o.is_object(), "{label}: each line is one object");
            }
        }

        // Values, not just well-formedness: comma placement can be wrong
        // in ways that still parse.
        let v = parse_all(&render_audit_entries_json(&[with_extras]));
        assert_eq!(v[0]["action"], "Deploy");
        assert_eq!(v[0]["when"], "2026-08-26T10:00:00Z");
        assert_eq!(v[0]["extras"]["bundle"], "app.zip");
        assert_eq!(v[0]["extras"]["size"], "12");

        // An entry with no extras must not carry an `extras` key at all.
        let bare = parse_all(&render_audit_entries_json(&[an_entry(Some("ok"), None)]));
        assert!(
            bare[0].get("extras").is_none(),
            "an empty extras map should be omitted, not emitted"
        );
    }

    #[test]
    fn escape_value_replaces_quotes_and_newlines() {
        assert_eq!(escape_value("plain"), "plain");
        assert_eq!(escape_value("with \"quotes\""), "with 'quotes'");
        assert_eq!(escape_value("line1\nline2"), "line1 line2");
        assert_eq!(escape_value("a\r\nb"), "a  b");
        assert_eq!(escape_value("a\tb"), "a b");
        assert_eq!(
            escape_value("AccessDenied: \"role\" not allowed\n  caused by: foo"),
            "AccessDenied: 'role' not allowed   caused by: foo"
        );
    }

    #[test]
    fn render_text_empty_says_no_entries() {
        let out = render_audit_entries_text(&[]);
        assert!(out.contains("no audit entries"));
    }

    #[test]
    fn render_text_columns_have_header_and_rows() {
        let entries = vec![parse_audit_line(
            "2026-05-27T10:15:30Z\taccount=A\tprofile=P\tregion=us-east-1\tstage=dispatched action=Restart target=my-env",
        )
        .unwrap()];
        let out = render_audit_entries_text(&entries);
        assert!(out.contains("TS"));
        assert!(out.contains("REGION"));
        assert!(out.contains("STAGE"));
        assert!(out.contains("ACTION"));
        assert!(out.contains("TARGET"));
        assert!(out.contains("OUTCOME"));
        assert!(out.contains("2026-05-27T10:15:30Z"));
        assert!(out.contains("us-east-1"));
        assert!(out.contains("dispatched"));
        assert!(out.contains("Restart"));
        assert!(out.contains("my-env"));
    }

    #[test]
    fn render_json_emits_jsonl() {
        let entries = vec![parse_audit_line(
            "2026-05-27T10:15:30Z\taccount=A\tprofile=P\tregion=us-east-1\tstage=dispatched action=Restart target=my-env",
        )
        .unwrap()];
        let out = render_audit_entries_json(&entries);
        // One JSON object per line.
        let lines: Vec<&str> = out.lines().collect();
        assert_eq!(lines.len(), 1);
        let line = lines[0];
        assert!(line.starts_with('{') && line.ends_with('}'));
        assert!(line.contains("\"when\":\"2026-05-27T10:15:30Z\""));
        assert!(line.contains("\"action\":\"Restart\""));
        assert!(line.contains("\"target\":\"my-env\""));
        // Absent fields (err, version, rule_id) should not appear.
        assert!(!line.contains("\"err\""));
        assert!(!line.contains("\"version\""));
    }

    #[test]
    fn render_json_escapes_quotes_and_control_chars() {
        let entries = vec![parse_audit_line(
            "2026-05-27T10:15:30Z\taccount=A\tprofile=P\tregion=r\tstage=completed action=Deploy target=env err=\"line1\\nline2 with \\\"quotes\\\"\"",
        )
        .unwrap()];
        let _out = render_audit_entries_json(&entries);
        // Just assert the function doesn't panic on tricky values.
        // (Round-trip semantics not in scope for v1; raw is the
        // source-of-truth log.)
    }

    /// **Golden pin** for `append_extras` — pins the exact wire shape
    /// of the `key=value` / `key="..."` encoding so a future quoting-
    /// policy change becomes a deliberate decision. Audit-log
    /// consumers (incident reviewers running `awk '$5 == "stage=…"'`)
    /// depend on this shape; silent changes invalidate their tooling.
    ///
    /// If this test fails: the change to `append_extras` is a wire-
    /// breaking change. Document the new format in the CHANGELOG,
    /// bump audit-shape version notes, and update this golden — or
    /// revert the change.
    ///
    /// Pinned in 0.19 (was a 0.18 review item).
    #[test]
    fn append_extras_golden_wire_shape() {
        let mut detail = String::from("stage=dispatched action=Demo target=env-1");
        append_extras(
            &mut detail,
            &[
                ("simple", "abc"),      // unquoted: no whitespace / quote / equals
                ("with_space", "a b"),  // quoted: contains whitespace
                ("with_quote", "a\"b"), // quoted + escaped
                ("with_equals", "a=b"), // quoted: contains '='
                ("empty", ""),          // quoted: empty value (distinguishable from omitted)
            ],
        );
        assert_eq!(
            detail,
            r#"stage=dispatched action=Demo target=env-1 simple=abc with_space="a b" with_quote="a'b" with_equals="a=b" empty="""#,
            "append_extras wire format changed — see test docstring before updating this constant"
        );
    }
}

#[cfg(test)]
mod parser_properties {
    use proptest::prelude::*;

    proptest! {
        #![proptest_config(ProptestConfig::with_cases(2000))]

        /// `parse_audit_line` must never panic, whatever is on the line.
        ///
        /// It is a hand-written char scanner with 13 `chars[i]` sites
        /// guarded by a hand-maintained `while i < n`, and it runs over
        /// `audit.log` — a file on disk that a crash or a full disk can
        /// leave truncated mid-write. Enumerating examples cannot cover
        /// that input space; this can.
        ///
        /// Scope, stated honestly: the parameter is `&str`, so a
        /// truncation that splits a UTF-8 sequence is rejected at the
        /// READ boundary and never reaches here. This covers malformed
        /// records, not malformed encoding.
        #[test]
        fn parse_audit_line_never_panics(s in "(?s).{0,400}") {
            let _ = super::parse_audit_line(&s);
        }

        /// The same, biased toward input that looks *nearly* valid —
        /// random text mostly misses the interesting branches, and the
        /// dangerous states are half-formed records, not noise.
        #[test]
        fn parse_audit_line_survives_near_miss_records(
            ts in "[0-9TZ:.-]{0,30}",
            verb in "[a-zA-Z]{0,20}",
            rest in "[a-zA-Z0-9=\"' \t\\\\-]{0,120}",
        ) {
            let _ = super::parse_audit_line(&format!("{ts}\t{verb}\t{rest}"));
            let _ = super::parse_audit_line(&format!("{ts}\t{verb}"));
            let _ = super::parse_audit_line(&ts);
        }

        /// `parse_kv_pairs` handles the quoting itself; unbalanced
        /// quotes and trailing escapes are exactly what a truncated
        /// write produces.
        #[test]
        fn parse_kv_pairs_never_panics(s in "[a-z0-9=\"'\\\\ \t]{0,200}") {
            let pairs = super::parse_kv_pairs(&s);
            // Whatever it returns, keys must be non-empty — an empty key
            // would make the audit line unreadable downstream.
            for (k, _) in pairs {
                prop_assert!(!k.is_empty());
            }
        }
    }

    /// A rollout audit line must record which profile the write went
    /// through.
    ///
    /// Rollout is the only CLI command that takes `--profile`, it is
    /// multi-region by construction, and `cli/action.rs` calls it "the
    /// biggest write the CLI has" — and its lines were the only ones
    /// carrying no account or profile, because this shape uses
    /// `rollout_id=` as its opener and went through the raw writer.
    /// "Which account did that rollout land in" is the first question
    /// the audit log is asked after an incident.
    #[test]
    fn a_rollout_line_records_the_profile() {
        let line = super::rollout_line(
            "rid-1",
            Some("prod"),
            "eu-west-1",
            "api",
            "v9",
            "dispatched",
            None,
        );
        assert!(
            line.contains("profile=prod"),
            "the profile the write went through must be on the line: {line}"
        );
        assert!(line.contains("rollout_id=rid-1"), "{line}");
        assert!(line.contains("region=eu-west-1"), "{line}");

        // Absent renders as `-`, matching the standard opener, rather
        // than vanishing and shifting the field order for a parser.
        let anon = super::rollout_line("rid-1", None, "eu-west-1", "api", "v9", "dispatched", None);
        assert!(anon.contains("profile=-"), "{anon}");
    }

    /// `AuditFilter` is what `ebman audit --env X --rule Y --action Z`
    /// resolves to, and it had no test at all — `cargo mutants` found
    /// both of its comparisons surviving.
    ///
    /// The `!=` flip is the one that matters: inverted, `--env api-prod`
    /// returns every entry EXCEPT api-prod's. An audit tool that
    /// confidently shows you the wrong subset is worse than one that
    /// errors, because you act on what it shows.
    ///
    /// The `<` on `since` is a boundary: an entry exactly ON the cutoff
    /// must be included, or `--since 1h` silently drops the oldest
    /// entry in its own window.
    #[test]
    fn the_audit_filter_selects_rather_than_excludes() {
        let entry = |target: &str, action: &str, when: &str| super::AuditEntry {
            when: when.to_string(),
            account: None,
            profile: None,
            region: None,
            rollout_id: None,
            stage: None,
            action: Some(action.to_string()),
            target: Some(target.to_string()),
            version: None,
            rule_id: None,
            outcome: None,
            err: None,
            extras: Default::default(),
            raw: String::new(),
        };
        let t0 = "2026-08-25T12:00:00Z";
        let api = entry("api-prod", "Deploy", t0);
        let worker = entry("worker-prod", "Restart", t0);

        // env: selects the named env, excludes the others.
        let f = super::AuditFilter {
            env: Some("api-prod"),
            ..Default::default()
        };
        assert!(f.matches(&api), "--env api-prod must MATCH api-prod");
        assert!(
            !f.matches(&worker),
            "--env api-prod must not match worker-prod — inverted, the tool \
             shows you everything except what you asked for"
        );

        // action: same shape, separate field.
        let f = super::AuditFilter {
            action: Some("Deploy"),
            ..Default::default()
        };
        assert!(f.matches(&api));
        assert!(!f.matches(&worker));

        // since: an entry exactly on the cutoff is IN the window.
        let cutoff = chrono::DateTime::parse_from_rfc3339(t0)
            .expect("fixed timestamp parses")
            .with_timezone(&chrono::Utc);
        let f = super::AuditFilter {
            since: Some(cutoff),
            ..Default::default()
        };
        assert!(
            f.matches(&api),
            "an entry exactly ON the --since cutoff must be included, or the \
             oldest entry in the requested window is silently dropped"
        );
        let older = entry("api-prod", "Deploy", "2026-08-25T11:59:59Z");
        assert!(!f.matches(&older), "and anything before it must not be");

        // An empty filter matches everything — otherwise the assertions
        // above could pass against a filter that rejects nothing.
        let all = super::AuditFilter::default();
        assert!(all.matches(&api) && all.matches(&worker));
    }

    /// Boundary cases in `parse_kv_pairs`'s unquoted-value lookahead.
    ///
    /// The existing tests cover the happy shapes; `cargo mutants` found
    /// eleven survivors in the lookahead that decides where an unquoted
    /// value ends. That lookahead is the whole difficulty of this
    /// format: values may contain spaces, so a space only terminates a
    /// value when what follows looks like `ident=`.
    ///
    /// It matters beyond display. `ebman audit replay` reconstructs an
    /// action from these fields and re-dispatches it, so a value that
    /// swallows the next key, or stops early, changes what gets replayed.
    #[test]
    fn unquoted_values_end_only_at_a_real_key_boundary() {
        let get = |s: &str, k: &str| -> Option<String> {
            super::parse_kv_pairs(s)
                .into_iter()
                .find(|(kk, _)| kk == k)
                .map(|(_, v)| v)
        };

        // A space followed by something that is NOT `ident=` stays in
        // the value. `=b` has a zero-length identifier before the `=`,
        // which is exactly the case the `j > ident_start` bound exists
        // for.
        assert_eq!(
            get("target=a =b stage=x", "target").as_deref(),
            Some("a =b"),
            "`=b` is not a key — a zero-length identifier must not end the value"
        );
        assert_eq!(get("target=a =b stage=x", "stage").as_deref(), Some("x"));

        // Repeated whitespace before a real key is still a boundary.
        assert_eq!(
            get("target=a  \t stage=x", "target").as_deref(),
            Some("a"),
            "the lookahead must skip ALL whitespace before testing for `ident=`"
        );
        assert_eq!(get("target=a  \t stage=x", "stage").as_deref(), Some("x"));

        // A value that runs to end-of-line keeps everything.
        assert_eq!(
            get("stage=dispatched target=env with spaces", "target").as_deref(),
            Some("env with spaces")
        );

        // Underscores and digits are identifier characters, so
        // `rollout_id=` terminates the previous value.
        assert_eq!(
            get("target=env-a rollout_id=r1", "target").as_deref(),
            Some("env-a")
        );
        assert_eq!(
            get("target=env-a rollout_id=r1", "rollout_id").as_deref(),
            Some("r1")
        );

        // A quoted value may contain both spaces and `=` without ending.
        assert_eq!(
            get(r#"err="denied: a=b c=d" stage=completed"#, "err").as_deref(),
            Some("denied: a=b c=d"),
            "quoting suspends the key-boundary lookahead entirely"
        );
        assert_eq!(
            get(r#"err="denied: a=b c=d" stage=completed"#, "stage").as_deref(),
            Some("completed")
        );
    }

    /// The module doc at the top of this file claims the writers and the
    /// parser are co-located so the line format has "a single source of
    /// truth". That claim went stale the moment `profile=` was added to
    /// rollout lines and the doc was not updated — which is exactly how
    /// the 0.34.2 review caught it.
    ///
    /// So the claim is now checked rather than asserted: every field the
    /// rollout writer emits must appear in the documented shape.
    #[test]
    fn the_documented_rollout_shape_matches_what_the_writer_emits() {
        let line = super::rollout_line(
            "rid",
            Some("prod"),
            "eu-west-1",
            "api",
            "v9",
            "dispatched",
            None,
        );
        let doc = include_str!("audit.rs");
        let doc_shape = doc
            .lines()
            .find(|l| l.contains("rollout_id=ID"))
            .expect("the module doc must describe the rollout line shape");

        for key in line
            .split(['\t', ' '])
            .filter_map(|tok| tok.split_once('=').map(|(k, _)| k))
            .filter(|k| !k.is_empty())
        {
            assert!(
                doc_shape.contains(&format!("{key}=")),
                "the writer emits `{key}=` but the module doc's rollout shape \
                 does not mention it — the 'single source of truth' comment is \
                 only true if this holds:\n  doc: {doc_shape}\n  line: {line}"
            );
        }
    }

    /// A profile name is operator-chosen free text, and these fields are
    /// space/tab separated — so a profile called `ops region=us-fake-1`
    /// could emit a second `region=` token that every consumer reads
    /// instead of the real one. `AuditFilter`'s `--region`, the text
    /// renderer, and `audit replay`'s region resolution all take the
    /// first match.
    ///
    /// Self-inflicted and pre-existing in the header opener, but 0.34.2
    /// added `profile=` to a NEW line shape, so it is fixed as a class:
    /// both openers now go through `field_token`, which quotes anything
    /// containing whitespace, `=` or `"`.
    #[test]
    fn a_profile_name_cannot_forge_another_field() {
        let hostile = "ops region=us-fake-1";
        let line = super::rollout_line(
            "rid",
            Some(hostile),
            "eu-west-1",
            "api",
            "v9",
            "dispatched",
            None,
        );

        let pairs = super::parse_kv_pairs(line.trim_start_matches('\t'));
        let regions: Vec<&String> = pairs
            .iter()
            .filter(|(k, _)| k == "region")
            .map(|(_, v)| v)
            .collect();
        assert_eq!(
            regions.len(),
            1,
            "a hostile profile name must not produce a second `region` key: \
             {line}\n  parsed: {pairs:?}"
        );
        assert_eq!(regions[0], "eu-west-1", "and the real region must win");

        let profile = pairs
            .iter()
            .find(|(k, _)| k == "profile")
            .map(|(_, v)| v.as_str());
        assert_eq!(
            profile,
            Some(hostile),
            "the whole profile name must survive as ONE value"
        );
    }
}

#[cfg(test)]
mod drain_tests {
    use super::{drain_webhooks, WEBHOOKS_IN_FLIGHT, WEBHOOK_COUNTER_LOCK};
    use std::sync::atomic::Ordering::SeqCst;
    use std::time::Duration;

    /// `drain_webhooks` waits for in-flight POSTs, and gives up at the
    /// deadline rather than hanging a one-shot CLI command forever.
    ///
    /// Five survivors sat on three expressions: the deadline
    /// arithmetic, the `> 0` in-flight test, and the `>= deadline`
    /// timeout. Time is paused, so `sleep` auto-advances and the
    /// elapsed measurements below are exact rather than wall-clock
    /// flaky.
    #[tokio::test(start_paused = true)]
    async fn nothing_in_flight_returns_immediately() {
        let _guard = WEBHOOK_COUNTER_LOCK.lock().await;
        WEBHOOKS_IN_FLIGHT.store(0, SeqCst);

        let start = tokio::time::Instant::now();
        drain_webhooks(Duration::from_secs(10)).await;
        assert_eq!(
            start.elapsed(),
            Duration::ZERO,
            "with nothing in flight the drain must not wait — `>= 0` here \
             would stall every one-shot command for the full timeout"
        );
    }

    /// And with something in flight it waits, then gives up at the
    /// deadline. Without this half, "return immediately" passes the
    /// case above and the drain never waits for anything.
    #[tokio::test(start_paused = true)]
    async fn an_in_flight_post_is_waited_for_up_to_the_deadline() {
        let _guard = WEBHOOK_COUNTER_LOCK.lock().await;
        WEBHOOKS_IN_FLIGHT.store(1, SeqCst);

        let start = tokio::time::Instant::now();
        drain_webhooks(Duration::from_millis(500)).await;
        let waited = start.elapsed();
        WEBHOOKS_IN_FLIGHT.store(0, SeqCst);

        assert!(
            waited >= Duration::from_millis(500),
            "the drain must wait for the POST — it returned after {waited:?}"
        );
        assert!(
            waited < Duration::from_secs(5),
            "and must give up at the deadline rather than hanging: {waited:?}"
        );
    }

    /// A POST that finishes mid-wait releases the drain early rather
    /// than holding it to the deadline.
    #[tokio::test(start_paused = true)]
    async fn a_finished_post_releases_the_drain_early() {
        let _guard = WEBHOOK_COUNTER_LOCK.lock().await;
        WEBHOOKS_IN_FLIGHT.store(1, SeqCst);

        tokio::spawn(async {
            tokio::time::sleep(Duration::from_millis(120)).await;
            WEBHOOKS_IN_FLIGHT.store(0, SeqCst);
        });

        let start = tokio::time::Instant::now();
        drain_webhooks(Duration::from_secs(30)).await;
        let waited = start.elapsed();
        WEBHOOKS_IN_FLIGHT.store(0, SeqCst);

        assert!(
            waited < Duration::from_secs(30),
            "the drain returned only at its deadline, so it is not \
             actually watching the counter: {waited:?}"
        );
    }

    /// A crafted action label must not be able to forge a `stage=`
    /// token.
    ///
    /// `action` carries operator-chosen text on several paths (a rename
    /// summary, a batch verb). `escape_value` does not quote spaces, so
    /// a bare interpolation let a value containing ` stage=completed`
    /// append a second `stage=` — and the parse loop lets the LAST
    /// duplicate win, so the forged value is the one every consumer
    /// reads. Self-forged, but it is the same forge path `field_token`
    /// exists to close for the header fields.
    #[test]
    fn a_crafted_action_label_cannot_forge_a_stage_token() {
        let target = "forge-probe-env";
        let path = crate::util::cache_dir().join("audit.log");
        let before = std::fs::read_to_string(&path).unwrap_or_default();

        crate::audit::append_action_refused(
            None,
            None,
            "eu-west-2",
            "Rename X stage=completed err=\"looks fine\"",
            target,
            "env_pinned",
            "clear the pin",
        );

        let after = std::fs::read_to_string(&path).unwrap_or_default();
        let delta = after
            .strip_prefix(&before)
            .expect("the audit log is append-only");
        let line = delta
            .lines()
            .find(|l| l.contains(target))
            .expect("the refusal was written");

        let entry = crate::audit::parse_audit_line(line).expect("the line must still parse");
        assert_eq!(
            entry.stage.as_deref(),
            Some("refused"),
            "a refusal must not be readable as a completed action: {line}"
        );
    }
}