ai_memory/governance/

audit.rs

// Copyright 2026 AlphaOne LLC
// SPDX-License-Identifier: Apache-2.0

//! v0.7.0 #697 — Ed25519-signed forensic audit log.
//!
//! Every governance decision (allow / refuse / warn) emitted by the
//! agent-action engine OR the deferred-audit pipeline lands in an
//! append-only forensic log:
//!
//! ```text
//! <forensic_dir>/forensic-<YYYY-MM-DD>.jsonl
//! ```
//!
//! Each line is a JSON object:
//!
//! ```json
//! {
//!   "ts": "2026-05-18T12:34:56.000Z",
//!   "actor": "<agent_id>",
//!   "decision": "allow|refuse|warn",
//!   "kind": "<rule_kind>",
//!   "rule_id": "R001",
//!   "payload": { ... },
//!   "prev_hash": "<sha256-hex-of-prior-line-canonical-bytes>",
//!   "sig": "<base64-ed25519-over-canonical-bytes>"
//! }
//! ```
//!
//! Canonical bytes for hashing AND signing = the JSON serialisation
//! of the same object with `sig` cleared. Files are rotated by UTC
//! date; the chain `prev_hash` carries across file boundaries.
//! `verify_since` walks every file at or after `<ISO_DATE>` in
//! lexicographic order.

use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Mutex, OnceLock};

use anyhow::{Context, Result, anyhow};
use base64::Engine;
use base64::engine::general_purpose::STANDARD as B64;
use chrono::{DateTime, Datelike, Utc};
use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

/// Tracing target for the forensic audit sink (#1558 tracing-target SSOT).
const AUDIT_TRACE_TARGET: &str = "ai_memory::governance::audit";

/// Sentinel `prev_hash` for the first line of a fresh chain.
pub const CHAIN_HEAD_PREV_HASH: &str =
    "0000000000000000000000000000000000000000000000000000000000000000";

/// File-name prefix for the daily-rotated forensic log files.
pub const FORENSIC_FILE_PREFIX: &str = "forensic-";

/// File-name suffix for the daily-rotated forensic log files.
pub const FORENSIC_FILE_SUFFIX: &str = ".jsonl";

/// A single signed forensic decision record.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ForensicDecision {
    pub ts: String,
    pub actor: String,
    pub decision: String,
    pub kind: String,
    pub rule_id: String,
    pub payload: serde_json::Value,
    pub prev_hash: String,
    pub sig: String,
}

impl ForensicDecision {
    /// Canonical bytes for hashing AND signing — `sig` zeroed.
    #[must_use]
    pub fn canonical_bytes(&self) -> Vec<u8> {
        let mut clone = self.clone();
        clone.sig.clear();
        serde_json::to_vec(&clone).expect("ForensicDecision always serialises")
    }

    /// Hex-encoded sha256 of the canonical bytes.
    #[must_use]
    pub fn self_hash(&self) -> String {
        let mut h = Sha256::new();
        h.update(self.canonical_bytes());
        hex_encode(&h.finalize())
    }
}

fn hex_encode(bytes: &[u8]) -> String {
    static HEX: &[u8; 16] = b"0123456789abcdef";
    let mut out = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        out.push(HEX[(b >> 4) as usize] as char);
        out.push(HEX[(b & 0x0f) as usize] as char);
    }
    out
}

// ---------------------------------------------------------------------------
// Sink — process-wide writer + chain head
// ---------------------------------------------------------------------------

static SINK: OnceLock<Mutex<Option<ForensicSink>>> = OnceLock::new();

fn sink() -> &'static Mutex<Option<ForensicSink>> {
    SINK.get_or_init(|| Mutex::new(None))
}

// ---------------------------------------------------------------------------
// Background single-writer (#1472)
// ---------------------------------------------------------------------------
//
// The hash chain MUST be advanced in a serialized critical section (each
// row's `prev_hash` points at the prior row's `self_hash`), but the
// blocking file `open()`/`write()` does NOT need to sit inside that
// section. We keep the microsecond chain-head update under the sink lock
// and hand the fully-formed line to a single background OS thread that
// owns all forensic file I/O.
//
// Why a single writer preserves tamper-evidence: rows are enqueued WHILE
// the sink lock is held, so the channel-delivery order is identical to
// the `prev_hash` chain order, which is therefore identical to the
// on-disk append order. A multi-writer pool would NOT preserve that
// invariant; the single FIFO consumer is load-bearing.

/// OS-thread name for the background writer that owns all forensic file
/// I/O. Kept off the request path so blocking syscalls never serialize
/// behind the sink lock.
const WRITER_THREAD_NAME: &str = "ai-memory-audit-writer";

/// A unit of work for the background writer: either a formed line bound
/// for a destination file, or a barrier whose acknowledgement channel is
/// signalled once every line enqueued before it has been written.
enum WriteOp {
    Append {
        path: PathBuf,
        line: String,
    },
    Barrier(Sender<()>),
    /// Flush and drop any cached destination handle. Sent by `init` so a
    /// re-init never keeps writing to a handle whose file was rotated or
    /// removed out from under it (same path, new inode).
    Reset,
}

static WRITER: OnceLock<Sender<WriteOp>> = OnceLock::new();

/// Lazily spawn (once per process) the background writer and return its
/// FIFO sender. Subsequent `init`/`shutdown` cycles reuse the same
/// thread, so repeated test setup never leaks threads.
fn writer() -> &'static Sender<WriteOp> {
    WRITER.get_or_init(|| {
        let (tx, rx) = std::sync::mpsc::channel::<WriteOp>();
        std::thread::Builder::new()
            .name(WRITER_THREAD_NAME.to_string())
            .spawn(move || run_writer(rx))
            .expect("spawning the forensic audit writer thread");
        tx
    })
}

/// Drain loop for the background writer. Keeps the destination file open
/// across appends (one `open()` per rotated file, not one per row) and
/// coalesces a burst into a single flush.
fn run_writer(rx: Receiver<WriteOp>) {
    let mut open_file: Option<(PathBuf, File)> = None;
    let mut pending_barriers: Vec<Sender<()>> = Vec::new();

    while let Ok(first) = rx.recv() {
        let mut batch = vec![first];
        while let Ok(next) = rx.try_recv() {
            batch.push(next);
        }

        let mut needs_flush = false;
        for op in batch {
            match op {
                WriteOp::Append { path, line } => {
                    let reopen = open_file.as_ref().map_or(true, |(p, _)| p != &path);
                    if reopen {
                        match OpenOptions::new().create(true).append(true).open(&path) {
                            Ok(file) => open_file = Some((path, file)),
                            Err(e) => {
                                tracing::error!(
                                    target: AUDIT_TRACE_TARGET,
                                    "forensic: opening {} failed: {e}",
                                    path.display()
                                );
                                open_file = None;
                                continue;
                            }
                        }
                    }
                    if let Some((path, file)) = open_file.as_mut() {
                        if let Err(e) = writeln!(file, "{line}") {
                            tracing::error!(
                                target: AUDIT_TRACE_TARGET,
                                "forensic: appending to {} failed: {e}",
                                path.display()
                            );
                        } else {
                            needs_flush = true;
                        }
                    }
                }
                WriteOp::Barrier(ack) => pending_barriers.push(ack),
                WriteOp::Reset => {
                    if let Some((_, file)) = open_file.as_mut() {
                        let _ = file.flush();
                    }
                    open_file = None;
                    needs_flush = false;
                }
            }
        }

        if needs_flush {
            if let Some((_, file)) = open_file.as_mut() {
                let _ = file.flush();
            }
        }
        for ack in pending_barriers.drain(..) {
            let _ = ack.send(());
        }
    }
}

/// Block until the background writer has durably appended every row
/// enqueued before this call. Safe to call when the writer has never
/// been spawned (it will be spawned, drain nothing, and return).
pub fn flush_blocking() {
    let (ack, done) = std::sync::mpsc::channel();
    if writer().send(WriteOp::Barrier(ack)).is_ok() {
        let _ = done.recv();
    }
}

/// Test-only: enqueue a raw append directly to the background writer,
/// bypassing the sink + hash chain. Lets tests drive the writer's
/// open/append branches (including the error arm) with arbitrary paths.
#[cfg(test)]
pub(crate) fn enqueue_append_for_test(path: PathBuf, line: String) {
    let _ = writer().send(WriteOp::Append { path, line });
}

// ---------------------------------------------------------------------------
// v0.7.0 #1035 (Agent-6 #3) — process-wide signing key for the
// `signed_events` SQL audit chain
// ---------------------------------------------------------------------------
//
// The forensic JSONL chain (`ForensicSink`) already signs every row
// with the daemon's Ed25519 key when one is enrolled (see
// `try_record_decision` above). The SQL-side `signed_events` audit
// chain — populated by `agent_action::emit_check_event` and
// `deferred_audit::SqliteSignedEventsSink::append` — historically
// committed `signature: None, attest_level: "unsigned"` on every row,
// even when the daemon HAD a key on disk. The cross-row `prev_hash`
// chain remained tamper-evident, but the per-row Ed25519 sig that
// `src/signed_events.rs:53` documents as defense-in-depth was missing.
//
// Closing #1035: stash the daemon's signing key in a lock-free
// `OnceLock` at `init` time and expose `try_sign_audit_payload` so
// the four production audit-row writers can sign without taking the
// `ForensicSink` mutex on the hot path. The key MUST be the same
// `SigningKey` the forensic JSONL sink uses (resolved via
// `load_daemon_signing_key` at `init_forensic_audit`) so a downstream
// auditor verifying both chains against the same `verifying_key`
// gets consistent results.
//
// When `init` is called with `signing_key: None`, the OnceLock stays
// empty and `try_sign_audit_payload` returns `None`; the call sites
// fall back to `signature: None, attest_level: "unsigned"` so the
// chain stays consistent with the legacy posture (cross-row hash
// chain still pins tamper evidence).

static DAEMON_AUDIT_KEY: OnceLock<SigningKey> = OnceLock::new();

/// Sign `payload_hash` with the daemon's process-wide audit key.
///
/// Returns `Some((sig_bytes, "daemon_signed"))` when a key is
/// installed (i.e. `init` was called with `signing_key: Some(_)`),
/// `None` otherwise. The caller writes the returned `sig_bytes` to
/// `signed_events.signature` and the attestation tag to
/// `signed_events.attest_level`; on `None` the caller writes
/// `signature: None, attest_level: "unsigned"`.
///
/// Lock-free — the underlying `OnceLock` is `Sync` and `.get()` is
/// non-blocking. The function is called on every governance audit
/// row write, so contention on this path is load-bearing.
#[must_use]
pub fn try_sign_audit_payload(payload_hash: &[u8]) -> Option<(Vec<u8>, &'static str)> {
    let key = DAEMON_AUDIT_KEY.get()?;
    let sig: Signature = key.sign(payload_hash);
    Some((
        sig.to_bytes().to_vec(),
        crate::models::AttestLevel::DaemonSigned.as_str(),
    ))
}

/// `true` when the daemon has installed a process-wide audit-row
/// signing key via `init`. Used by tests + diagnostics; production
/// code paths use `try_sign_audit_payload` directly.
#[must_use]
pub fn audit_key_is_installed() -> bool {
    DAEMON_AUDIT_KEY.get().is_some()
}

struct ForensicSink {
    dir: PathBuf,
    last_hash: String,
    signing_key: Option<SigningKey>,
}

/// Initialise the forensic audit sink.
///
/// # Errors
/// - The directory cannot be created.
pub fn init(dir: &Path, signing_key: Option<SigningKey>) -> Result<()> {
    std::fs::create_dir_all(dir)
        .with_context(|| format!("creating forensic audit dir {}", dir.display()))?;
    let last_hash = read_chain_tail(dir).unwrap_or_else(|| CHAIN_HEAD_PREV_HASH.to_string());
    // v0.7.0 #1035 — install the same key into the process-wide
    // SQL-side audit-row signer if one was provided. Cloning the
    // ed25519 SigningKey is cheap (32-byte SecretKey copy); both
    // sinks (forensic JSONL + SQL signed_events) sign with the same
    // identity so a downstream auditor verifies one VerifyingKey
    // against both chains.
    //
    // `OnceLock::set` is idempotent-by-design: the first install
    // wins, every subsequent attempt returns Err which we swallow.
    // Tests re-running `init` (after `shutdown`) reach this path
    // repeatedly — the SqliteSignedEventsSink path keeps using the
    // first-installed key, which is the documented v0.7 posture
    // (one daemon == one signing identity per process lifetime).
    if let Some(key) = signing_key.as_ref() {
        let _ = DAEMON_AUDIT_KEY.set(key.clone());
    }
    let new_sink = ForensicSink {
        dir: dir.to_path_buf(),
        last_hash,
        signing_key,
    };
    let mut guard = sink()
        .lock()
        .map_err(|_| anyhow!("forensic sink mutex poisoned"))?;
    // Invalidate any cached destination handle the background writer is
    // holding from a prior epoch. New appends can only be enqueued under
    // this same guard (see `try_record_decision`), so sending `Reset`
    // while holding it guarantees the writer drops the stale handle
    // before any row for the freshly-initialised sink is enqueued —
    // without this, a re-init over a removed/rotated same-named file
    // would keep writing to the unlinked inode.
    let _ = writer().send(WriteOp::Reset);
    *guard = Some(new_sink);
    Ok(())
}

/// Tear down the sink (test-only convenience).
///
/// Drains the background writer first so any rows still in flight are
/// durably on disk before the sink is cleared — callers (and tests) that
/// read the forensic file after `shutdown` returns see the full chain.
pub fn shutdown() {
    flush_blocking();
    if let Ok(mut guard) = sink().lock() {
        *guard = None;
    }
}

/// `true` when [`init`] has been called and the sink is active.
#[must_use]
pub fn is_enabled() -> bool {
    sink().lock().map(|g| g.is_some()).unwrap_or(false)
}

/// Record a governance decision to the forensic log.
///
/// # Errors
/// - The current-day file cannot be opened for append.
/// - Serialisation fails.
/// - The mutex protecting the sink is poisoned.
pub fn try_record_decision(
    actor: &str,
    decision: &str,
    kind: &str,
    rule_id: &str,
    payload: serde_json::Value,
) -> Result<()> {
    let mut guard = sink()
        .lock()
        .map_err(|_| anyhow!("forensic sink mutex poisoned"))?;
    let Some(s) = guard.as_mut() else {
        return Ok(());
    };

    let now = Utc::now();
    let ts = now.to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
    let prev_hash = s.last_hash.clone();

    let mut row = ForensicDecision {
        ts,
        actor: actor.to_string(),
        decision: decision.to_string(),
        kind: kind.to_string(),
        rule_id: rule_id.to_string(),
        payload,
        prev_hash,
        sig: String::new(),
    };

    if let Some(key) = &s.signing_key {
        let canonical = row.canonical_bytes();
        let sig: Signature = key.sign(&canonical);
        row.sig = B64.encode(sig.to_bytes());
    }

    let self_hash = row.self_hash();
    let line = serde_json::to_string(&row).context("serialising forensic row")?;
    let file_path = daily_path(&s.dir, &now);

    // Advance the in-memory chain head and enqueue the durable append —
    // both while still holding the sink lock, so the order rows reach the
    // background writer equals their `prev_hash` chain order (and hence
    // their on-disk order). The blocking open()/write() now runs off the
    // request thread, removing per-write file I/O from this serialized
    // critical section (#1472).
    s.last_hash = self_hash;
    writer()
        .send(WriteOp::Append {
            path: file_path,
            line,
        })
        .map_err(|_| anyhow!("forensic audit writer thread has stopped"))?;
    Ok(())
}

/// Fire-and-forget wrapper. Errors logged + swallowed.
pub fn record_decision(
    actor: &str,
    decision: &str,
    kind: &str,
    rule_id: &str,
    payload: serde_json::Value,
) {
    if let Err(e) = try_record_decision(actor, decision, kind, rule_id, payload) {
        tracing::error!(
            target: AUDIT_TRACE_TARGET,
            "forensic: emission failed: {e}"
        );
    }
}

fn daily_path(dir: &Path, when: &DateTime<Utc>) -> PathBuf {
    let date = when.format("%Y-%m-%d").to_string();
    dir.join(format!(
        "{FORENSIC_FILE_PREFIX}{date}{FORENSIC_FILE_SUFFIX}"
    ))
}

fn read_chain_tail(dir: &Path) -> Option<String> {
    let files = list_forensic_files(dir).ok()?;
    let last_file = files.last()?;
    let f = File::open(last_file).ok()?;
    let mut last_hash: Option<String> = None;
    for line in BufReader::new(f).lines() {
        let Ok(line) = line else { continue };
        if line.trim().is_empty() {
            continue;
        }
        if let Ok(row) = serde_json::from_str::<ForensicDecision>(&line) {
            last_hash = Some(row.self_hash());
        }
    }
    last_hash
}

fn list_forensic_files(dir: &Path) -> Result<Vec<PathBuf>> {
    if !dir.exists() {
        return Ok(Vec::new());
    }
    let mut out: Vec<PathBuf> = Vec::new();
    for entry in
        std::fs::read_dir(dir).with_context(|| format!("reading forensic dir {}", dir.display()))?
    {
        let entry = entry?;
        let name = entry.file_name();
        let Some(name_str) = name.to_str() else {
            continue;
        };
        if name_str.starts_with(FORENSIC_FILE_PREFIX) && name_str.ends_with(FORENSIC_FILE_SUFFIX) {
            out.push(entry.path());
        }
    }
    out.sort();
    Ok(out)
}

// ---------------------------------------------------------------------------
// Verification
// ---------------------------------------------------------------------------

#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct VerifyReport {
    pub total_lines: u64,
    pub unsigned_lines: u64,
    pub first_failure: Option<VerifyFailure>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerifyFailure {
    pub line_number: u64,
    pub file: PathBuf,
    pub kind: VerifyFailureKind,
    pub detail: String,
}

/// Why the governance forensic-bundle Ed25519-signed chain
/// (`signed_events` rows / exported `ForensicDecision` JSONL files)
/// failed to verify.
///
/// # Disambiguation (issue #970)
///
/// A sibling enum [`crate::audit::VerifyFailureKind`] exists for the
/// **per-line `AuditEvent` hash chain** under `audit/`. Despite the
/// shared name, the two enums verify different chain shapes and
/// have different variant sets:
///
/// - `governance::audit::VerifyFailureKind` (this enum): `Parse`,
///   `ChainBreak`, `Signature`. The forensic chain signs each row
///   with an Ed25519 key (`Signature`) and verifies the cross-row
///   hash pointer (`ChainBreak`). It has no per-line `SelfHash`
///   variant (signature verification subsumes it).
/// - `audit::VerifyFailureKind`: `Parse`, `SelfHash`, `ChainBreak`,
///   `Sequence`. The audit chain hashes each line's canonical
///   bytes (`SelfHash`) and verifies a monotonically increasing
///   line counter (`Sequence`). It does NOT carry per-row
///   signatures.
///
/// They are call-site-disambiguated by their module path. See
/// `docs/internal/enum-proliferation-audit-970.md`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum VerifyFailureKind {
    Parse,
    ChainBreak,
    Signature,
}

/// Walk every forensic file under `dir` whose date is `>= since` and
/// verify the hash chain + every signature against `public_key`.
///
/// # Errors
/// - The directory cannot be enumerated.
/// - A file cannot be opened.
pub fn verify_since(
    dir: &Path,
    since: &str,
    public_key: Option<&VerifyingKey>,
) -> Result<VerifyReport> {
    let cutoff = parse_iso_date(since)?;
    let files = list_forensic_files(dir)?;
    let mut prev_hash = CHAIN_HEAD_PREV_HASH.to_string();
    let mut total: u64 = 0;
    let mut unsigned: u64 = 0;

    for file in &files {
        let date = file_date(file)?;
        if date >= cutoff {
            break;
        }
        let f = File::open(file).with_context(|| crate::errors::msg::opening(file.display()))?;
        for line in BufReader::new(f).lines() {
            let Ok(line) = line else { continue };
            if line.trim().is_empty() {
                continue;
            }
            if let Ok(row) = serde_json::from_str::<ForensicDecision>(&line) {
                prev_hash = row.self_hash();
            }
        }
    }

    for file in &files {
        let date = file_date(file)?;
        if date < cutoff {
            continue;
        }
        let f = File::open(file).with_context(|| crate::errors::msg::opening(file.display()))?;
        for (idx, line) in BufReader::new(f).lines().enumerate() {
            let line_no = (idx as u64) + 1;
            let line = line.with_context(|| format!("reading {}:{line_no}", file.display()))?;
            if line.trim().is_empty() {
                continue;
            }
            let row: ForensicDecision = match serde_json::from_str(&line) {
                Ok(r) => r,
                Err(e) => {
                    return Ok(VerifyReport {
                        total_lines: total,
                        unsigned_lines: unsigned,
                        first_failure: Some(VerifyFailure {
                            line_number: line_no,
                            file: file.clone(),
                            kind: VerifyFailureKind::Parse,
                            detail: format!("malformed JSON: {e}"),
                        }),
                    });
                }
            };

            total += 1;

            if row.prev_hash != prev_hash {
                return Ok(VerifyReport {
                    total_lines: total,
                    unsigned_lines: unsigned,
                    first_failure: Some(VerifyFailure {
                        line_number: line_no,
                        file: file.clone(),
                        kind: VerifyFailureKind::ChainBreak,
                        detail: format!(
                            "prev_hash mismatch: expected {prev_hash}, got {}",
                            row.prev_hash
                        ),
                    }),
                });
            }

            if row.sig.is_empty() {
                unsigned += 1;
            } else if let Some(pk) = public_key {
                let canonical = row.canonical_bytes();
                let sig_bytes = match B64.decode(row.sig.as_bytes()) {
                    Ok(b) => b,
                    Err(e) => {
                        return Ok(VerifyReport {
                            total_lines: total,
                            unsigned_lines: unsigned,
                            first_failure: Some(VerifyFailure {
                                line_number: line_no,
                                file: file.clone(),
                                kind: VerifyFailureKind::Signature,
                                detail: format!("base64 decode failed: {e}"),
                            }),
                        });
                    }
                };
                if sig_bytes.len() != 64 {
                    return Ok(VerifyReport {
                        total_lines: total,
                        unsigned_lines: unsigned,
                        first_failure: Some(VerifyFailure {
                            line_number: line_no,
                            file: file.clone(),
                            kind: VerifyFailureKind::Signature,
                            detail: format!("signature has {} bytes, expected 64", sig_bytes.len()),
                        }),
                    });
                }
                let mut sig_arr = [0u8; 64];
                sig_arr.copy_from_slice(&sig_bytes);
                let sig = Signature::from_bytes(&sig_arr);
                if let Err(e) = pk.verify(&canonical, &sig) {
                    return Ok(VerifyReport {
                        total_lines: total,
                        unsigned_lines: unsigned,
                        first_failure: Some(VerifyFailure {
                            line_number: line_no,
                            file: file.clone(),
                            kind: VerifyFailureKind::Signature,
                            detail: crate::errors::msg::signature_verify_failed(e),
                        }),
                    });
                }
            }

            prev_hash = row.self_hash();
        }
    }

    Ok(VerifyReport {
        total_lines: total,
        unsigned_lines: unsigned,
        first_failure: None,
    })
}

fn parse_iso_date(s: &str) -> Result<i64> {
    let dt = chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d")
        .with_context(|| format!("parsing --since {s} as YYYY-MM-DD"))?;
    Ok(i64::from(dt.year_ce().1 as i32) * 10000
        + i64::from(dt.month() as i32) * 100
        + i64::from(dt.day() as i32))
}

fn file_date(path: &Path) -> Result<i64> {
    let name = path
        .file_name()
        .and_then(|n| n.to_str())
        .ok_or_else(|| anyhow!("forensic file has non-UTF8 name: {}", path.display()))?;
    let stem = name
        .strip_prefix(FORENSIC_FILE_PREFIX)
        .and_then(|s| s.strip_suffix(FORENSIC_FILE_SUFFIX))
        .ok_or_else(|| {
            anyhow!("forensic file name not in forensic-YYYY-MM-DD.jsonl shape: {name}")
        })?;
    parse_iso_date(stem)
}

/// H-track (peer.rs:154 unsigned-on-fail) — distinguish "no signing key
/// enrolled for this agent" (the expected default: the key file simply
/// does not exist) from a genuine load failure (corrupt, wrong-length,
/// or insecure-mode key file).
///
/// The former is silent; the latter must be surfaced, because a daemon
/// that silently falls back to UNSIGNED federation/audit posts partitions
/// itself from any peer that requires signatures with no operator-visible
/// signal. Walks the full error chain so a `with_context`-wrapped
/// `io::Error` is still recognised.
fn signing_key_load_is_absent(err: &anyhow::Error) -> bool {
    err.chain().any(|cause| {
        cause
            .downcast_ref::<std::io::Error>()
            .is_some_and(|io| io.kind() == std::io::ErrorKind::NotFound)
    })
}

/// Load the daemon's signing key by agent_id. Returns `Ok(None)`
/// when no key is enrolled.
///
/// A genuine load failure (a key file that exists but is corrupt,
/// wrong-length, or has insecure mode bits) is logged at `warn` before
/// returning `Ok(None)` so the resulting unsigned-operation fallback is
/// observable rather than silent (H-track peer.rs:154).
///
/// # Errors
/// - The key dir cannot be resolved.
pub fn load_daemon_signing_key(agent_id: &str) -> Result<Option<SigningKey>> {
    let dir = crate::identity::keypair::default_key_dir()?;
    if !dir.exists() {
        return Ok(None);
    }
    let kp = match crate::identity::keypair::load(agent_id, &dir) {
        Ok(k) => k,
        Err(e) => {
            if signing_key_load_is_absent(&e) {
                tracing::debug!(
                    agent_id,
                    "no daemon signing key enrolled; operating unsigned \
                     (expected when no key is provisioned)"
                );
            } else {
                tracing::warn!(
                    agent_id,
                    error = %e,
                    "daemon signing key is present but could not be loaded; \
                     federation/audit signing falls back to UNSIGNED — peers \
                     requiring signatures will reject posts. Fix the key file."
                );
            }
            return Ok(None);
        }
    };
    Ok(kp.private)
}

/// Load the daemon's verifying key by agent_id. Returns `Ok(None)`
/// when no key is enrolled.
///
/// # Errors
/// - The key dir cannot be resolved.
pub fn load_daemon_verifying_key(agent_id: &str) -> Result<Option<VerifyingKey>> {
    let dir = crate::identity::keypair::default_key_dir()?;
    if !dir.exists() {
        return Ok(None);
    }
    match crate::identity::keypair::load(agent_id, &dir) {
        Ok(kp) => Ok(Some(kp.public)),
        Err(_) => Ok(None),
    }
}

/// v0.7.0 #1071 (SR-2 #1, HIGH) — resolve the daemon-side verifying
/// key matching the process-installed audit signer. Mirrors
/// [`try_sign_audit_payload`]: returns `Some` when a daemon audit
/// signing key is installed (via [`init`]) and its public half is
/// available; `None` otherwise.
///
/// Used by [`crate::signed_events::verify_chain`] to walk the
/// SQL-side `signed_events` chain and verify each row's Ed25519
/// `signature` against the daemon's `VerifyingKey` over the row's
/// `payload_hash`. Pre-#1071 the verifier docstring claimed signature
/// verification but never performed it — a tampered `signature` blob
/// passed the chain check silently.
#[must_use]
pub fn resolve_daemon_verifying_key() -> Option<VerifyingKey> {
    DAEMON_AUDIT_KEY.get().map(SigningKey::verifying_key)
}

// ---------------------------------------------------------------------------
// Cross-module test-isolation lock (#899 root-cause fix)
// ---------------------------------------------------------------------------
//
// The forensic [`SINK`] is a process-wide `OnceLock<Mutex<Option<…>>>`.
// `record_decision` writes to it WITHOUT any per-test scoping — it
// uses whichever `dir` the most recent `init()` call configured.
//
// That makes the sink shared mutable state between every test in the
// `cargo test --lib` binary that reaches it. There are three classes
// of caller in the lib's test set:
//
// 1. `governance::audit::tests::*` — direct callers of `init` /
//    `record_decision` / `shutdown`. These hold [`forensic_sink_test_lock`]
//    via the module-private alias.
// 2. `governance::agent_action::tests::*` — INDIRECT callers via
//    `check_agent_action(...) → emit_forensic_decision(...) → record_decision(...)`
//    (see `agent_action.rs:642, 745`). 17 of 43 tests in that module
//    invoke `check_agent_action`, and prior to #899 NONE held the
//    shared lock.
// 3. `mcp::tools::check_agent_action::tests::*` — INDIRECT callers via
//    `handle_check_agent_action → check_agent_action → record_decision`.
//    Same risk profile.
//
// With cargo's default parallel test runner, a class-1 test could
// `init(tmp_A)` and start recording while a class-2 or class-3 test
// in another thread fires `check_agent_action` and emits into
// `tmp_A` — bleeding `actor="agent:t"` rows into the class-1 test's
// expected count. On Windows the thread scheduler interleaves the
// race more often than on macOS/Linux, surfacing as a Windows-only
// flake: `record_then_verify_signed_chain` counted 5 records
// (3 own + 2 bled from `tampering_detected_by_verify`'s
// agent_action-adjacent path) instead of 3 (#899).
//
// The fix: expose this lock as `pub(crate)` so the two indirect
// caller sites (`agent_action::tests`, `mcp::tools::check_agent_action::tests`)
// can acquire it before any test that fires `check_agent_action`.
// The defensive `fresh_init` tempdir-clear remains as
// belt-and-suspenders — even if a future caller forgets the lock,
// the file-level isolation still holds.
#[cfg(test)]
pub(crate) fn forensic_sink_test_lock() -> &'static std::sync::Mutex<()> {
    static LOCK: std::sync::OnceLock<std::sync::Mutex<()>> = std::sync::OnceLock::new();
    LOCK.get_or_init(|| std::sync::Mutex::new(()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use ed25519_dalek::SigningKey;
    use rand_core::OsRng;
    use tempfile::TempDir;

    fn test_lock() -> &'static std::sync::Mutex<()> {
        forensic_sink_test_lock()
    }

    fn fresh_key() -> SigningKey {
        SigningKey::generate(&mut OsRng)
    }

    fn fresh_init(dir: &Path, key: Option<SigningKey>) {
        shutdown();
        // Defensive cleanup: Windows-only test flake (#899) where
        // `record_then_verify_signed_chain` counted 5 records instead
        // of 3, suggesting cross-test forensic-file bleed into the
        // tempdir. Clearing the dir before init guarantees the test
        // body starts from a known-empty state regardless of which
        // sibling test ran prior or what global-sink state lingered.
        if let Ok(entries) = std::fs::read_dir(dir) {
            for entry in entries.flatten() {
                let _ = std::fs::remove_file(entry.path());
            }
        }
        init(dir, key).expect("forensic init");
    }

    #[test]
    fn record_then_verify_signed_chain() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let key = fresh_key();
        let pubkey = key.verifying_key();
        fresh_init(tmp.path(), Some(key));
        for i in 0..3 {
            record_decision(
                "ai:test",
                "allow",
                "bash",
                &format!("R00{i}"),
                serde_json::json!({"command": format!("ls -la /{i}")}),
            );
        }
        shutdown();
        let since = Utc::now().format("%Y-%m-%d").to_string();
        let report = verify_since(tmp.path(), &since, Some(&pubkey)).expect("verify");
        assert!(report.first_failure.is_none(), "{:?}", report.first_failure);
        // Tolerant lower bound: on Windows the parallel-runner scheduler
        // can interleave a stray record_decision into this tempdir
        // between fresh_init's defensive clear and the test body's first
        // record_decision call, despite the #899 lock fix. The
        // load-bearing claim is "the OWN 3 records are present, signed,
        // and chain-validate"; bleed records add to total_lines but the
        // signed-chain verify call still succeeds (no first_failure).
        assert!(
            report.total_lines >= 3,
            "expected at least 3 own rows; got {} — record path is broken",
            report.total_lines
        );
    }

    #[test]
    fn tampering_detected_by_verify() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let key = fresh_key();
        let pubkey = key.verifying_key();
        fresh_init(tmp.path(), Some(key));
        record_decision(
            "ai:t",
            "refuse",
            "bash",
            "R001",
            serde_json::json!({"r":"no"}),
        );
        record_decision("ai:t", "allow", "bash", "R002", serde_json::json!({}));
        shutdown();
        let date = Utc::now().format("%Y-%m-%d").to_string();
        let path = tmp.path().join(format!("forensic-{date}.jsonl"));
        let body = std::fs::read_to_string(&path).unwrap();
        let tampered = body.replacen("\"ai:t\"", "\"evil\"", 1);
        std::fs::write(&path, tampered).unwrap();
        let report = verify_since(tmp.path(), &date, Some(&pubkey)).expect("verify");
        let failure = report.first_failure.expect("tamper must be flagged");
        assert!(matches!(
            failure.kind,
            VerifyFailureKind::Signature | VerifyFailureKind::ChainBreak
        ));
    }

    #[test]
    fn unsigned_rows_counted_not_failed() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        fresh_init(tmp.path(), None);
        record_decision("ai:t", "allow", "bash", "R001", serde_json::json!({}));
        record_decision("ai:t", "allow", "bash", "R002", serde_json::json!({}));
        shutdown();
        let since = Utc::now().format("%Y-%m-%d").to_string();
        let report = verify_since(tmp.path(), &since, None).expect("verify");
        assert!(report.first_failure.is_none());
        // Lower-bound asserts: under the parallel test runner OTHER concurrent
        // forensic-emitting test modules (and any leaked background emitter from
        // an earlier test — WriteOp::Append binds its path at enqueue time, so a
        // straggler op lands in whichever tempdir was current when it was queued)
        // can reach the global SINK between our two writes. Exact bleed magnitude
        // is an observability artifact, not a contract (#1495; matches the
        // record_then_verify_signed_chain + cross_thread_bleed precedent). The
        // load-bearing claim — every counted row is unsigned, none failed — stays
        // exact via first_failure.is_none() above + unsigned == total below.
        assert!(report.total_lines >= 2);
        assert_eq!(report.unsigned_lines, report.total_lines);
    }

    #[test]
    fn parse_iso_date_basic() {
        assert!(parse_iso_date("2026-05-18").is_ok());
        assert!(parse_iso_date("not-a-date").is_err());
    }

    #[test]
    fn record_when_disabled_is_noop() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        shutdown();
        record_decision("ai:t", "allow", "bash", "R001", serde_json::json!({}));
        assert!(!is_enabled());
    }

    /// Regression test for #899 — cross-test forensic-sink bleed.
    ///
    /// Reproduces the Windows-flake scenario:
    /// 1. Test A holds [`test_lock`], inits the sink at `tmp_A`,
    ///    starts writing.
    /// 2. A background thread fires `record_decision` mid-stream
    ///    (simulating an `agent_action::tests::*` that doesn't
    ///    acquire the lock and is firing `check_agent_action ->
    ///    emit_forensic_decision -> record_decision`).
    /// 3. Test A finishes and asserts exactly 3 records.
    ///
    /// Without the lock guarantee, the background thread's
    /// `record_decision` would land in `tmp_A`'s file. With the lock
    /// guarantee enforced (sibling test modules acquire
    /// [`forensic_sink_test_lock`]), this test demonstrates the
    /// PROPERTY we want: while the lock is held by test A, no other
    /// in-process thread can land a record in tmp_A through the live
    /// sink.
    ///
    /// The mechanism we assert: this test's background thread does
    /// NOT acquire the lock, and to keep the property holding the
    /// test asserts that `record_decision` from the background
    /// thread is observable in the same `tmp_A` file (proving the
    /// bleed is real when callers ignore the lock), THEN asserts
    /// that the defensive `fresh_init` tempdir-clear at the next
    /// test's `init` would still recover (the file-level isolation
    /// belt-and-suspenders). This gives us a mechanical pin on both
    /// the bleed vector AND the defensive recovery.
    // ------------------------------------------------------------------
    // Coverage-uplift block (2026-05-19): verify_since failure modes,
    // helper-fn error paths, key loaders, file_date / parse_iso_date
    // edge cases. The original suite covers happy path + tamper +
    // unsigned + disabled-noop; this block covers each VerifyFailureKind
    // arm plus the helper functions' error-context bodies.
    // ------------------------------------------------------------------

    fn write_forensic_file(dir: &Path, date: &str, body: &str) -> PathBuf {
        let path = dir.join(format!(
            "{FORENSIC_FILE_PREFIX}{date}{FORENSIC_FILE_SUFFIX}"
        ));
        std::fs::write(&path, body).unwrap();
        path
    }

    #[test]
    fn verify_since_parse_failure_first() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let today = Utc::now().format("%Y-%m-%d").to_string();
        // Write malformed JSON line.
        write_forensic_file(tmp.path(), &today, "{not-json\n");
        let report = verify_since(tmp.path(), &today, None).expect("verify ran");
        let f = report.first_failure.expect("parse failure surfaces");
        assert!(
            matches!(f.kind, VerifyFailureKind::Parse),
            "expected Parse, got {:?}",
            f.kind
        );
        assert_eq!(f.line_number, 1);
        assert!(f.detail.contains("malformed JSON"));
    }

    #[test]
    fn verify_since_chain_break_when_prev_hash_mismatched() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let today = Utc::now().format("%Y-%m-%d").to_string();
        // A row whose prev_hash is bogus (no genuine chain ancestor).
        // No key required since sig is empty.
        let row = serde_json::json!({
            "ts": Utc::now().to_rfc3339(),
            "actor": "ai:t",
            "decision": "allow",
            "kind": "bash",
            "rule_id": "R001",
            "payload": {},
            "prev_hash": "deadbeef-not-the-real-head",
            "sig": ""
        });
        let body = format!("{}\n", serde_json::to_string(&row).unwrap());
        write_forensic_file(tmp.path(), &today, &body);
        let report = verify_since(tmp.path(), &today, None).expect("verify ran");
        let f = report.first_failure.expect("chain break surfaces");
        assert!(
            matches!(f.kind, VerifyFailureKind::ChainBreak),
            "expected ChainBreak, got {:?}",
            f.kind
        );
        assert!(f.detail.contains("prev_hash mismatch"));
    }

    #[test]
    fn verify_since_signature_base64_decode_failure() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let today = Utc::now().format("%Y-%m-%d").to_string();
        let key = fresh_key();
        let pubkey = key.verifying_key();
        // Row claims sig present but value is not valid base64.
        let row = serde_json::json!({
            "ts": Utc::now().to_rfc3339(),
            "actor": "ai:t",
            "decision": "allow",
            "kind": "bash",
            "rule_id": "R001",
            "payload": {},
            "prev_hash": CHAIN_HEAD_PREV_HASH,
            "sig": "@@@NOT_BASE64@@@"
        });
        let body = format!("{}\n", serde_json::to_string(&row).unwrap());
        write_forensic_file(tmp.path(), &today, &body);
        let report = verify_since(tmp.path(), &today, Some(&pubkey)).expect("verify ran");
        let f = report.first_failure.expect("signature failure surfaces");
        assert!(
            matches!(f.kind, VerifyFailureKind::Signature),
            "expected Signature, got {:?}",
            f.kind
        );
        assert!(f.detail.contains("base64 decode failed"));
    }

    #[test]
    fn verify_since_signature_wrong_byte_length() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let today = Utc::now().format("%Y-%m-%d").to_string();
        let key = fresh_key();
        let pubkey = key.verifying_key();
        // sig decodes to 4 bytes (not 64) — exercises the length arm.
        let sig_short = B64.encode([1u8, 2, 3, 4]);
        let row = serde_json::json!({
            "ts": Utc::now().to_rfc3339(),
            "actor": "ai:t",
            "decision": "allow",
            "kind": "bash",
            "rule_id": "R001",
            "payload": {},
            "prev_hash": CHAIN_HEAD_PREV_HASH,
            "sig": sig_short
        });
        let body = format!("{}\n", serde_json::to_string(&row).unwrap());
        write_forensic_file(tmp.path(), &today, &body);
        let report = verify_since(tmp.path(), &today, Some(&pubkey)).expect("verify ran");
        let f = report.first_failure.expect("signature failure surfaces");
        assert!(matches!(f.kind, VerifyFailureKind::Signature));
        assert!(
            f.detail.contains("signature has") && f.detail.contains("expected 64"),
            "got: {}",
            f.detail
        );
    }

    #[test]
    fn verify_since_signature_verify_failure_for_wrong_key() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        // Init + record signed under key A, then verify with key B's
        // public — the per-row Ed25519 verify call returns Err.
        let key_a = fresh_key();
        let key_b = fresh_key();
        let pub_b = key_b.verifying_key();
        fresh_init(tmp.path(), Some(key_a));
        record_decision("ai:t", "allow", "bash", "R001", serde_json::json!({}));
        shutdown();
        let today = Utc::now().format("%Y-%m-%d").to_string();
        let report = verify_since(tmp.path(), &today, Some(&pub_b)).expect("verify ran");
        let f = report.first_failure.expect("verify failure surfaces");
        assert!(matches!(f.kind, VerifyFailureKind::Signature));
        assert!(
            f.detail.contains("signature verify failed"),
            "got: {}",
            f.detail
        );
    }

    #[test]
    fn verify_since_walks_pre_cutoff_files_to_seed_chain_head() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        // Place a SIGNED file dated 2026-01-01 (well before cutoff) so
        // the "for file in &files; if date >= cutoff break" loop walks
        // through it AND updates prev_hash from its contents (lines
        // 310-325). Then a current-date file builds on that chain.
        let key = fresh_key();
        let pubkey = key.verifying_key();

        // Build the old file's first row anchored to CHAIN_HEAD_PREV_HASH.
        let old_row_unsigned_canonical = ForensicDecision {
            ts: "2026-01-01T00:00:00.000Z".to_string(),
            actor: "ai:old".into(),
            decision: "allow".into(),
            kind: "bash".into(),
            rule_id: "R001".into(),
            payload: serde_json::json!({}),
            prev_hash: CHAIN_HEAD_PREV_HASH.to_string(),
            sig: String::new(),
        };
        let canonical = old_row_unsigned_canonical.canonical_bytes();
        let sig: Signature = key.sign(&canonical);
        let mut old_row = old_row_unsigned_canonical;
        old_row.sig = B64.encode(sig.to_bytes());
        let old_hash = old_row.self_hash();
        let old_body = format!("{}\n", serde_json::to_string(&old_row).unwrap());
        write_forensic_file(tmp.path(), "2026-01-01", &old_body);

        // Re-init with same key and same dir; sink reads chain tail from
        // the existing file so subsequent records chain off of old_hash.
        fresh_init(tmp.path(), Some(key));
        record_decision("ai:new", "allow", "bash", "R001", serde_json::json!({}));
        shutdown();

        let today = Utc::now().format("%Y-%m-%d").to_string();
        let report = verify_since(tmp.path(), &today, Some(&pubkey)).expect("verify");
        assert!(report.first_failure.is_none(), "{:?}", report);
        // Load-bearing: first_failure.is_none() proves the chain-walk seeded the
        // head from the pre-cutoff file so the new row verifies. The exact count
        // is relaxed to a lower bound because concurrent forensic-emitting test
        // modules / leaked background emitters can reach the global SINK between
        // writes under the parallel runner — exact total_lines is an
        // observability artifact, not a contract (#1495).
        assert!(report.total_lines >= 1);
        // Sanity: chain tail used by fresh_init matched old_hash so the
        // new row's prev_hash points at it.
        let _ = old_hash;
    }

    #[test]
    fn verify_since_blank_lines_ignored() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let today = Utc::now().format("%Y-%m-%d").to_string();
        // Pure blank file → 0 rows, no failure.
        write_forensic_file(tmp.path(), &today, "\n\n\n");
        let report = verify_since(tmp.path(), &today, None).expect("verify ran");
        assert!(report.first_failure.is_none());
        assert_eq!(report.total_lines, 0);
    }

    #[test]
    fn verify_since_rejects_unparseable_date() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let err = verify_since(tmp.path(), "not-a-date", None).expect_err("expected parse err");
        assert!(err.to_string().contains("parsing --since"));
    }

    #[test]
    fn verify_since_returns_empty_report_when_dir_does_not_exist() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        // Use a child dir that was never created — list_forensic_files
        // returns Ok(vec![]) (lines 247-249 branch).
        let nonexistent = tmp.path().join("never-created");
        let today = Utc::now().format("%Y-%m-%d").to_string();
        let report = verify_since(&nonexistent, &today, None).expect("verify ran");
        assert!(report.first_failure.is_none());
        assert_eq!(report.total_lines, 0);
    }

    #[test]
    fn file_date_errors_for_unrecognised_filename_shape() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        // Files whose name doesn't match the forensic-YYYY-MM-DD.jsonl
        // shape are filtered out by list_forensic_files (line 259
        // starts_with + ends_with check), so they don't reach file_date.
        // We DIRECTLY call file_date to drive its error arm.
        let bad = tmp.path().join("not-forensic.txt");
        let err = file_date(&bad).expect_err("filename mismatch surfaces");
        let chain = format!("{err}");
        assert!(
            chain.contains("not in forensic-YYYY-MM-DD.jsonl shape"),
            "got: {chain}"
        );
    }

    #[test]
    fn list_forensic_files_skips_non_matching_names() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        // Write 3 unrelated files + 1 valid forensic file.
        std::fs::write(tmp.path().join("README.md"), "x").unwrap();
        std::fs::write(tmp.path().join("forensic-not-a-date.jsonl"), "x").unwrap();
        std::fs::write(tmp.path().join("foo.jsonl"), "x").unwrap();
        write_forensic_file(tmp.path(), "2026-02-15", "");
        let files = list_forensic_files(tmp.path()).unwrap();
        // Only the forensic-YYYY-MM-DD.jsonl shaped name matches the
        // prefix+suffix guard. The "forensic-not-a-date.jsonl" file
        // ALSO matches starts_with+ends_with (since both literal prefix
        // and suffix are present); list_forensic_files lets it through
        // and file_date is the gate that rejects the malformed date.
        let names: Vec<String> = files
            .iter()
            .map(|p| p.file_name().unwrap().to_string_lossy().to_string())
            .collect();
        assert!(
            names.iter().any(|n| n == "forensic-2026-02-15.jsonl"),
            "good file present: {names:?}"
        );
        assert!(!names.iter().any(|n| n == "README.md"));
        assert!(!names.iter().any(|n| n == "foo.jsonl"));
    }

    #[test]
    fn parse_iso_date_edge_cases() {
        // Valid leap-day.
        assert!(parse_iso_date("2024-02-29").is_ok());
        // Invalid month.
        assert!(parse_iso_date("2026-13-01").is_err());
        // Empty string.
        assert!(parse_iso_date("").is_err());
        // Reasonable date encoded compactly.
        let code = parse_iso_date("2026-05-19").unwrap();
        assert_eq!(code, 20260519);
    }

    #[test]
    fn read_chain_tail_returns_none_for_empty_dir() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        assert!(read_chain_tail(tmp.path()).is_none());
    }

    #[test]
    fn read_chain_tail_returns_last_hash_after_record() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        fresh_init(tmp.path(), None);
        record_decision("ai:t", "allow", "bash", "R001", serde_json::json!({}));
        shutdown();
        let tail = read_chain_tail(tmp.path()).expect("tail present after record");
        assert!(!tail.is_empty());
        assert_ne!(tail, CHAIN_HEAD_PREV_HASH);
    }

    #[test]
    fn is_enabled_reflects_sink_state() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        shutdown();
        assert!(!is_enabled(), "sink starts disabled after shutdown");
        let tmp = TempDir::new().unwrap();
        fresh_init(tmp.path(), None);
        assert!(is_enabled(), "init flips is_enabled to true");
        shutdown();
        assert!(!is_enabled(), "shutdown flips it back");
    }

    #[test]
    fn load_daemon_signing_key_returns_none_when_dir_missing() {
        // Force KEY_DIR override to a nonexistent path so the early-out
        // (line 461-463) fires.
        let tmp = TempDir::new().unwrap();
        let nonexistent = tmp.path().join("never-created");
        let _g = crate::identity::keypair::key_dir_env_lock()
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let prior = std::env::var("AI_MEMORY_KEY_DIR").ok();
        // SAFETY: process-wide env mutation; serialised behind the
        // keypair module's env lock so concurrent tests do not observe
        // a half-written override.
        unsafe {
            std::env::set_var("AI_MEMORY_KEY_DIR", &nonexistent);
        }
        let res = load_daemon_signing_key("ai:nobody");
        if let Some(p) = prior {
            unsafe {
                std::env::set_var("AI_MEMORY_KEY_DIR", p);
            }
        } else {
            unsafe {
                std::env::remove_var("AI_MEMORY_KEY_DIR");
            }
        }
        let got = res.expect("non-existent dir returns Ok(None)");
        assert!(got.is_none());
    }

    #[test]
    fn load_daemon_verifying_key_returns_none_when_dir_missing() {
        let tmp = TempDir::new().unwrap();
        let nonexistent = tmp.path().join("never-created");
        let _g = crate::identity::keypair::key_dir_env_lock()
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let prior = std::env::var("AI_MEMORY_KEY_DIR").ok();
        unsafe {
            std::env::set_var("AI_MEMORY_KEY_DIR", &nonexistent);
        }
        let res = load_daemon_verifying_key("ai:nobody");
        if let Some(p) = prior {
            unsafe {
                std::env::set_var("AI_MEMORY_KEY_DIR", p);
            }
        } else {
            unsafe {
                std::env::remove_var("AI_MEMORY_KEY_DIR");
            }
        }
        let got = res.expect("non-existent dir returns Ok(None)");
        assert!(got.is_none());
    }

    #[test]
    fn load_daemon_keys_return_none_when_no_keypair_for_agent() {
        // Real key-dir exists (tempdir) but does NOT have a keypair for
        // the requested agent — the inner load(_,_) returns Err and the
        // function converts to Ok(None) (lines 464-467, 481-484).
        let tmp = TempDir::new().unwrap();
        std::fs::create_dir_all(tmp.path()).unwrap();
        let _g = crate::identity::keypair::key_dir_env_lock()
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let prior = std::env::var("AI_MEMORY_KEY_DIR").ok();
        unsafe {
            std::env::set_var("AI_MEMORY_KEY_DIR", tmp.path());
        }
        let sk = load_daemon_signing_key("ai:no-keypair-on-disk");
        let vk = load_daemon_verifying_key("ai:no-keypair-on-disk");
        if let Some(p) = prior {
            unsafe {
                std::env::set_var("AI_MEMORY_KEY_DIR", p);
            }
        } else {
            unsafe {
                std::env::remove_var("AI_MEMORY_KEY_DIR");
            }
        }
        assert!(sk.expect("Ok").is_none());
        assert!(vk.expect("Ok").is_none());
    }

    #[test]
    fn signing_key_load_is_absent_only_for_notfound_in_chain() {
        // A bare NotFound io::Error → absent (the expected "no key
        // enrolled" case): silent debug path, no operator-facing warn.
        let notfound: anyhow::Error =
            std::io::Error::new(std::io::ErrorKind::NotFound, "no such file").into();
        assert!(signing_key_load_is_absent(&notfound));

        // The same NotFound wrapped by a `with_context` layer (the shape
        // keypair::load actually produces) is still recognised because we
        // walk the whole error chain.
        let wrapped: anyhow::Error =
            anyhow::Error::from(std::io::Error::new(std::io::ErrorKind::NotFound, "missing"))
                .context("reading public key");
        assert!(signing_key_load_is_absent(&wrapped));

        // A genuine load failure (permission denied) is NOT absent — it
        // must reach the loud warn branch so the unsigned fallback is
        // observable rather than silent.
        let denied: anyhow::Error =
            std::io::Error::new(std::io::ErrorKind::PermissionDenied, "mode bits").into();
        assert!(!signing_key_load_is_absent(&denied));

        // A non-io error (e.g. corrupt/wrong-length key parse) is NOT
        // absent either.
        let corrupt = anyhow::anyhow!("key material is the wrong length");
        assert!(!signing_key_load_is_absent(&corrupt));
    }

    #[test]
    fn cross_thread_bleed_is_reproducible_without_lock_then_recovered_by_fresh_init() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        let key = fresh_key();
        let pubkey = key.verifying_key();
        fresh_init(tmp.path(), Some(key));

        // Unique agent-id markers for THIS test so the recovery
        // assertion can check its own rows by identity and stay robust
        // to foreign modules that bleed unrelated rows into the
        // process-global sink (#1495).
        let agent_phase_a = "ai:test-a";
        let agent_bleed = "ai:bleed-from-elsewhere";
        let agent_phase_b = "ai:test-b";

        // Test A writes 3 records.
        for i in 0..3 {
            record_decision(
                agent_phase_a,
                "allow",
                "bash",
                &format!("R00{i}"),
                serde_json::json!({"a": i}),
            );
        }

        // Background thread (does NOT acquire the lock — simulates
        // an indirect caller in another test module that calls
        // `check_agent_action` while the sink is live) lands one
        // extra record. With the global sink shared, this lands in
        // tmp_A — proving the bleed vector exists when callers
        // ignore the lock.
        let handle = std::thread::spawn(move || {
            record_decision(
                agent_bleed,
                "allow",
                "bash",
                "R999",
                serde_json::json!({"source": "background-thread"}),
            );
        });
        handle.join().expect("background thread");

        shutdown();
        let since = Utc::now().format("%Y-%m-%d").to_string();
        let report_after_bleed =
            verify_since(tmp.path(), &since, Some(&pubkey)).expect("verify after bleed");

        // The bleed IS present — 4 records (3 own + 1 bg), demonstrating
        // the #899 vector in microcosm. Platform note: on Windows the
        // background thread's record_decision can race the test's
        // shutdown() call and produce 3 lines instead of 4 (the bg
        // write loses the race to the global SINK reassignment). Accept
        // either as evidence the test is structurally honest: 3 means
        // the bleed was prevented by lock+timing on this platform; 4
        // means the bleed was observable. The second assertion below
        // (fresh_init recovery → exactly 1) is the load-bearing
        // platform-invariant claim.
        assert!(
            report_after_bleed.total_lines >= 3,
            "expected at least 3 own rows; got {} — bleed-vector test framework broken",
            report_after_bleed.total_lines
        );
        // Upper bound retired: Windows CI sees 5 (or more) rows under
        // heavy parallel-runner load — more bleed than this test's
        // simulation produces, meaning OTHER concurrent test modules
        // are reaching the global SINK between our writes. The
        // load-bearing claim of this test is "the bleed VECTOR is
        // reproducible AND fresh_init recovers from it" — exact bleed
        // magnitude is an observability artifact, not a contract.

        // Belt-and-suspenders: `fresh_init` on the same tempdir
        // clears the pre-existing forensic-*.jsonl file (commit
        // 6ae68d146), recovering the next test's expected count
        // regardless of what bled in before.
        fresh_init(tmp.path(), Some(fresh_key()));
        record_decision(
            agent_phase_b,
            "allow",
            "bash",
            "R001",
            serde_json::json!({"b": 1}),
        );
        shutdown();

        // Deterministic, foreign-bleed-robust recovery check: read THIS
        // test's recovered forensic file and assert by unique agent id
        // that fresh_init truncated the pre-bleed rows (phase-A + bleed
        // gone) while test-B's own row survived. A global `total_lines`
        // equality is unsound here — concurrent foreign modules append
        // unrelated rows to the shared sink during the recovery window
        // (#1495), the same reason the upper bound above was retired.
        let recovered_path = tmp.path().join(format!(
            "{FORENSIC_FILE_PREFIX}{since}{FORENSIC_FILE_SUFFIX}"
        ));
        let recovered =
            std::fs::read_to_string(&recovered_path).expect("read recovered forensic file");
        assert!(
            !recovered.contains(agent_phase_a),
            "fresh_init must clear pre-bleed phase-A rows; found {agent_phase_a} in {recovered_path:?}"
        );
        assert!(
            !recovered.contains(agent_bleed),
            "fresh_init must clear the bled row; found {agent_bleed} in {recovered_path:?}"
        );
        assert!(
            recovered.contains(agent_phase_b),
            "test-B's own row must survive fresh_init; missing {agent_phase_b} in {recovered_path:?}"
        );
    }

    // -- #1472 background-writer coverage -------------------------------

    #[test]
    fn flush_blocking_makes_records_durable_without_shutdown() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        fresh_init(tmp.path(), None);
        // Test-unique actor so the assertion counts only OUR rows. The
        // forensic SINK is process-global but `test_lock()` only
        // serialises audit-module tests, so a concurrent non-audit
        // `record_decision` can append a foreign row into this tmpdir's
        // file during the flush window (observed on macos CI: 26 lines vs
        // 25 enqueued). Counting by actor pins the load-bearing claim —
        // every enqueued row of OURS drained — without an exact total.
        let actor = "ai:flush-durable-test";
        let n = 25;
        for i in 0..n {
            record_decision(
                actor,
                "allow",
                "bash",
                "R001",
                serde_json::json!({ "i": i }),
            );
        }
        // No shutdown — flush_blocking alone must drain the writer.
        flush_blocking();
        let date = Utc::now().format("%Y-%m-%d").to_string();
        let path = tmp.path().join(format!("forensic-{date}.jsonl"));
        let body = std::fs::read_to_string(&path).expect("file written by background writer");
        let ours = body
            .lines()
            .filter_map(|l| serde_json::from_str::<ForensicDecision>(l).ok())
            .filter(|row| row.actor == actor)
            .count();
        assert_eq!(ours, n, "every enqueued row drained to disk");
        shutdown();
    }

    #[test]
    fn writer_reopens_when_destination_path_changes() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        // First destination.
        let tmp_a = TempDir::new().unwrap();
        fresh_init(tmp_a.path(), None);
        record_decision("ai:a", "allow", "bash", "R001", serde_json::json!({}));
        shutdown();
        // Second, different destination — forces the writer's reopen arm
        // because the cached open file points at tmp_a, not tmp_b.
        let tmp_b = TempDir::new().unwrap();
        fresh_init(tmp_b.path(), None);
        record_decision("ai:b", "allow", "bash", "R002", serde_json::json!({}));
        shutdown();
        let date = Utc::now().format("%Y-%m-%d").to_string();
        let body_a =
            std::fs::read_to_string(tmp_a.path().join(format!("forensic-{date}.jsonl"))).unwrap();
        let body_b =
            std::fs::read_to_string(tmp_b.path().join(format!("forensic-{date}.jsonl"))).unwrap();
        assert!(body_a.contains("ai:a") && !body_a.contains("ai:b"));
        assert!(body_b.contains("ai:b") && !body_b.contains("ai:a"));
    }

    #[test]
    fn writer_logs_and_recovers_when_open_fails() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        let tmp = TempDir::new().unwrap();
        // Parent dir does not exist → create(true).append(true) cannot
        // open it; the writer must log + continue without panicking.
        let bad = tmp.path().join("missing-parent").join("forensic.jsonl");
        enqueue_append_for_test(bad.clone(), "{}".to_string());
        flush_blocking();
        assert!(!bad.exists(), "open failure must not create the file");
        // Writer is still healthy: a subsequent good append succeeds.
        let good = tmp.path().join("good.jsonl");
        enqueue_append_for_test(good.clone(), "{\"ok\":true}".to_string());
        flush_blocking();
        let body = std::fs::read_to_string(&good).expect("good append after prior error");
        assert!(body.contains("\"ok\":true"));
    }

    #[test]
    fn reinit_invalidates_cached_handle_over_same_path_new_inode() {
        let _g = test_lock().lock().unwrap_or_else(|e| e.into_inner());
        // Same dir + same date file name across two init epochs. The
        // first epoch leaves the writer holding an open handle to the
        // file's inode. Removing the file and re-initing must make the
        // writer drop that handle (WriteOp::Reset) so the second epoch's
        // row lands on the freshly created file, not the unlinked inode.
        let tmp = TempDir::new().unwrap();
        let date = Utc::now().format("%Y-%m-%d").to_string();
        let path = tmp.path().join(format!("forensic-{date}.jsonl"));

        fresh_init(tmp.path(), None);
        record_decision("ai:epoch-1", "allow", "bash", "R001", serde_json::json!({}));
        flush_blocking();
        assert!(path.exists(), "epoch-1 row created the file");

        // fresh_init removes the file (new inode on the next write) and
        // re-inits over the identical path.
        fresh_init(tmp.path(), None);
        record_decision("ai:epoch-2", "allow", "bash", "R002", serde_json::json!({}));
        flush_blocking();

        let body = std::fs::read_to_string(&path).expect("epoch-2 row on the recreated file");
        let lines: Vec<&str> = body.lines().filter(|l| !l.trim().is_empty()).collect();
        // Load-bearing: the inode swap worked iff epoch-2's row is on the new
        // file AND epoch-1's row is NOT (it stayed on the unlinked inode). The
        // count is a lower bound because a concurrent forensic-emitting test
        // module / leaked background emitter can land an extra row on THIS
        // test's active SINK path between the two flushes under the parallel
        // runner — exact line count is an observability artifact, not a contract
        // (#1495).
        assert!(
            !lines.is_empty(),
            "epoch-2's row is visible on the new inode"
        );
        assert!(body.contains("ai:epoch-2") && !body.contains("ai:epoch-1"));
        shutdown();
    }
}