cortex_ledger/

jsonl.rs

//! Append-only JSONL event log: `JsonlLog`.
//!
//! Per BUILD_SPEC §7, the JSONL mirror is a peer to the SQLite store: it
//! provides **inspectability** (one event per line, grep-able) and
//! **disaster recovery** (the full ledger can be rebuilt from this file
//! alone). Append-only, fsync-per-write, and hash-chained.
//!
//! ## Append protocol
//!
//! 1. The caller hands [`JsonlLog::append`] an [`Event`].
//! 2. The log sets `prev_event_hash` to its current head (or `None` for
//!    the first event).
//! 3. The log re-seals the event via [`crate::hash::seal`], which
//!    recomputes both `payload_hash` and `event_hash` under the canonical
//!    framing.
//! 4. The log writes one JSON line + `\n`, then **fsyncs the file**.
//! 5. The new `event_hash` becomes the head.
//!
//! Re-sealing on append means callers don't need to pre-populate the
//! hashes (they're an artifact of the chain, not the event's identity).
//!
//! ## Why fsync per write
//!
//! The JSONL log is the disaster-recovery source of truth. An event that
//! is "appended" but lost on power loss leaves the SQL store ahead of the
//! mirror — defeating the mirror's purpose. We pay the latency cost
//! (~1ms-10ms per write on commodity SSDs) in exchange for crash safety.
//! Higher-throughput modes (group commit, periodic fsync) are a future
//! optimization gated on a config flag and an ADR.
//!
//! ## What this module does NOT do
//!
//! - Replicate the chain to remote storage (out of scope; future ADR).
//! - Compact or rotate the log (out of scope for v0; planned for Phase 4).
//! - Index the log (the SQL store is the queryable surface).

use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

use chrono::Utc;
use cortex_core::{
    attestor::Attestor, canonical::canonical_signing_input, schema_migration_v1_to_v2_event, Event,
    EventSource, PolicyDecision, PolicyOutcome, SchemaMigrationV1ToV2Payload,
};
use thiserror::Error;

use crate::anchor_chain::{row_preimage, GENESIS_PREV_SIGNATURE};
use crate::hash::seal;
use crate::signed_row::{b64_decode, b64_encode, RowSignature, SignedRow};

/// Required contributor rule id documenting that the event source tier gate
/// composed into the policy decision for an unsigned JSONL append
/// (ADR 0019 §3, ADR 0026 §2). The ledger refuses
/// [`EventSource::User`] rows when the final outcome is
/// [`PolicyOutcome::Reject`] or [`PolicyOutcome::Quarantine`].
pub const APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID: &str = "ledger.append.event_source_tier_gate";
/// Required contributor rule id documenting that attestation requirements
/// (ADR 0010 §1, ADR 0014 §3, ADR 0026 §4) composed into the policy
/// decision for an unsigned JSONL append. The contributor MUST vote
/// `Allow` for `EventSource::User`; the ledger refuses authority-bearing
/// rows that lack attestation.
pub const APPEND_ATTESTATION_REQUIRED_RULE_ID: &str = "ledger.append.attestation_required";
/// Required contributor rule id documenting that the runtime mode gate
/// (ADR 0037 §2) composed into the policy decision for an unsigned JSONL
/// append. Local-development unsigned ledgers register a `Warn`; trusted
/// modes register `Reject` to prevent unsigned rows from being passed off
/// as authority grade.
pub const APPEND_RUNTIME_MODE_RULE_ID: &str = "ledger.append.runtime_mode";
/// Required contributor rule id documenting that the signing key state at
/// event time satisfies ADR 0023 current-use revalidation for a signed
/// JSONL append. Historical-only or revoked keys vote `Reject` here.
pub const APPEND_SIGNED_KEY_STATE_CURRENT_USE_RULE_ID: &str =
    "ledger.append_signed.key_state_current_use";
/// Required contributor rule id documenting that the signing principal's
/// trust tier satisfies the ADR 0019 minimum for a signed JSONL append.
/// Principals below `Verified` vote `Reject` here.
pub const APPEND_SIGNED_TRUST_TIER_MINIMUM_RULE_ID: &str =
    "ledger.append_signed.trust_tier_minimum";
/// Required contributor rule id documenting that the proposing principal sits
/// in the `Operator` authority class (ADR 0019 §3) for a v1 -> v2 schema
/// migration boundary append. Non-operator principals vote `Reject` here; the
/// rule is documented as authority-class so a future ADR 0019 §7 scoped
/// `tier_admin` capability can satisfy the same contributor.
pub const SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID: &str =
    "ledger.schema_migration.authority_class";
/// Required contributor rule id documenting that a fresh operator attestation
/// (ADR 0010 §1-§2) was supplied over the proposed v1 -> v2 boundary payload.
/// Absent or invalid attestation votes `Reject`. ADR 0026 §4 forbids
/// `BreakGlass` substituting for this contributor at the migration authority
/// root.
pub const SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID: &str =
    "ledger.schema_migration.attestation_required";
/// Required contributor rule id documenting that the signing key supplied for
/// the operator attestation is in current use (ADR 0023 §2 / §5): the key
/// state at attestation time is `Active`, not `Retired` or `Revoked`. A
/// historical-only signing key votes `Reject` here; ADR 0026 §4 forbids
/// `BreakGlass` substituting for this contributor.
pub const SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID: &str =
    "ledger.schema_migration.current_use_temporal_authority";

/// Errors raised by [`JsonlLog`].
#[derive(Debug, Error)]
pub enum JsonlError {
    /// File-system I/O failure.
    #[error("io error on {path:?}: {source}")]
    Io {
        /// Path that was being read or written.
        path: PathBuf,
        /// Underlying I/O error.
        #[source]
        source: std::io::Error,
    },
    /// A JSON decode failed during iteration / verification.
    #[error("json decode error at line {line} in {path:?}: {source}")]
    Decode {
        /// Path being read.
        path: PathBuf,
        /// 1-based line number.
        line: usize,
        /// Underlying serde error.
        #[source]
        source: serde_json::Error,
    },
    /// A JSON encode failed during append.
    #[error("json encode error: {0}")]
    Encode(#[source] serde_json::Error),
    /// Chain verification failed (one or more rows broke the chain).
    ///
    /// Callers needing per-row diagnostics should use
    /// [`crate::audit::verify_chain`] instead.
    #[error("chain verification failed: {0}")]
    ChainBroken(String),
    /// Validation failed before append.
    #[error("validation failed: {0}")]
    Validation(String),
}

/// Append-only JSONL log handle.
///
/// Owns the file path and an in-memory copy of the current chain head.
/// The file handle is opened per write to keep the type `Send + Sync` and
/// to make file-rotation / handoff straightforward (if we ever need it).
///
/// Per Lane 3.D.6 / ADR 0010 §1-§2 the log also tracks the **previous
/// row's Ed25519 signature bytes** so the next signed append can build
/// the canonical preimage (which couples `S_n` to `S_{n-1}`).
#[derive(Debug)]
pub struct JsonlLog {
    path: PathBuf,
    /// Hex `event_hash` of the most-recently-appended event.
    head: Option<String>,
    /// Number of rows currently in the file.
    len: u64,
    /// 32-byte truncation of the most-recently-appended row's Ed25519
    /// signature (or all-zero genesis sentinel for an empty log). Used
    /// as `S_{n-1}` input to [`row_preimage`] on the next signed append.
    /// We carry only the first 32 bytes because that is what the
    /// canonical lineage hex captures (Ed25519 sigs are 64 bytes; the
    /// canonical preimage hashes the lineage hex string, so the chosen
    /// truncation is what binds rows together — see ADR 0010 §2 Option A
    /// "prev_signature digest inside `P_n`").
    last_sig_prefix: [u8; 32],
}

/// Logical identifier (`ledger_id`) for the JSONL chain. Used as part of
/// every row's canonical preimage so signatures from one log cannot be
/// replayed into another.
///
/// Exposed `pub(crate)` so [`crate::audit::verify_signed_chain`] derives
/// the same id from the same path on the verify side.
pub(crate) fn ledger_id_for(path: &Path) -> String {
    path.file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("cortex-jsonl")
        .to_string()
}

impl JsonlLog {
    /// Open (or create) the log at `path`.
    ///
    /// On open we scan the existing file to recover the head hash and row
    /// count. An empty file (or non-existent path) is a fresh log with no
    /// head.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, JsonlError> {
        let path = path.as_ref().to_path_buf();
        // Touch the file so subsequent appends succeed without a TOCTOU.
        OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .map_err(|source| JsonlError::Io {
                path: path.clone(),
                source,
            })?;

        // Scan to recover head, len, and last signature prefix.
        let f = File::open(&path).map_err(|source| JsonlError::Io {
            path: path.clone(),
            source,
        })?;
        let reader = BufReader::new(f);
        let mut head: Option<String> = None;
        let mut len: u64 = 0;
        let mut last_sig_prefix: [u8; 32] = GENESIS_PREV_SIGNATURE;
        for (i, line) in reader.lines().enumerate() {
            let line = line.map_err(|source| JsonlError::Io {
                path: path.clone(),
                source,
            })?;
            if line.trim().is_empty() {
                continue;
            }
            let row: SignedRow =
                serde_json::from_str(&line).map_err(|source| JsonlError::Decode {
                    path: path.clone(),
                    line: i + 1,
                    source,
                })?;
            head = Some(row.event.event_hash.clone());
            len += 1;
            if let Some(sig) = &row.signature {
                if let Some(bytes) = b64_decode(&sig.bytes) {
                    if bytes.len() >= 32 {
                        last_sig_prefix.copy_from_slice(&bytes[..32]);
                    } else {
                        // Malformed-but-present signature on disk; leave
                        // last_sig_prefix at the genesis sentinel and let
                        // the audit verifier surface it as BadSignature.
                    }
                }
            } else {
                // Legacy unsigned row: do NOT advance last_sig_prefix.
                // The audit verifier will flag this row's MissingSignature
                // failure; subsequent signed appends will still chain off
                // whatever prior signature was last seen (or genesis).
            }
        }

        Ok(Self {
            path,
            head,
            len,
            last_sig_prefix,
        })
    }

    /// Path of the underlying file.
    #[must_use]
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Number of rows currently in the log.
    #[must_use]
    pub fn len(&self) -> u64 {
        self.len
    }

    /// Whether the log has zero rows.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    /// Current chain head (`event_hash` of the most recent row).
    ///
    /// `None` for an empty log.
    #[must_use]
    pub fn head(&self) -> Option<&str> {
        self.head.as_deref()
    }

    /// Append one event to the log **without** an Ed25519 signature, gated
    /// through the ADR 0026 enforcement lattice.
    ///
    /// This is the legacy / pre-3.D.6 path. Rows written via this method
    /// **fail** [`crate::audit::verify_signed_chain`] with
    /// [`crate::audit::FailureReason::MissingSignature`] (per ADR 0010 §1
    /// "Single asymmetric trust domain": rows without a valid Ed25519
    /// signature do not verify; there is no symmetric-MAC fallback). It
    /// is retained for the local-development ledger path and for tests of
    /// non-attestation features (hash chain, payload framing);
    /// **trusted-history / authority paths MUST use
    /// [`Self::append_signed`]**.
    ///
    /// `policy` is the composed [`PolicyDecision`] for this append and
    /// MUST satisfy:
    ///
    /// 1. The composition includes contributors for
    ///    [`APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID`],
    ///    [`APPEND_ATTESTATION_REQUIRED_RULE_ID`], and
    ///    [`APPEND_RUNTIME_MODE_RULE_ID`]. Callers that skipped composition
    ///    are refused.
    /// 2. When `event.source` is [`EventSource::User`], the attestation
    ///    contributor MUST vote [`PolicyOutcome::Allow`] regardless of
    ///    final outcome. ADR 0026 §4 forbids `BreakGlass` from substituting
    ///    for the attestation requirement at the user-event authority
    ///    boundary.
    /// 3. A final outcome of [`PolicyOutcome::Reject`] or
    ///    [`PolicyOutcome::Quarantine`] fails closed and writes nothing.
    pub fn append(
        &mut self,
        mut event: Event,
        policy: &PolicyDecision,
    ) -> Result<String, JsonlError> {
        require_append_contributors(policy)?;
        require_event_source_attestation(policy, &event.source)?;
        require_append_final_outcome(policy, "ledger.append")?;

        event.prev_event_hash = self.head.clone();
        seal(&mut event);
        let row = SignedRow::unsigned(event);
        let line = serde_json::to_string(&row).map_err(JsonlError::Encode)?;
        self.write_line(&line)?;
        self.head = Some(row.event.event_hash.clone());
        self.len += 1;
        // Unsigned append does NOT advance last_sig_prefix.
        Ok(row.event.event_hash)
    }

    /// Append one event with an Ed25519 signature over the canonical
    /// attestation preimage (T-3.D.6, ADR 0010 §1-§2), gated through the
    /// ADR 0026 enforcement lattice.
    ///
    /// The preimage couples this row's signature to `S_{n-1}` (the
    /// previous row's signature, or the genesis sentinel for the first
    /// row), the row's `event_id`, `payload_hash`, `session_id`,
    /// `ledger_id`, and the signing key's `key_id`. See
    /// [`row_preimage`] for the exact shape.
    ///
    /// `policy` is the composed [`PolicyDecision`] for this signed append
    /// and MUST satisfy:
    ///
    /// 1. The composition includes contributors for
    ///    [`APPEND_SIGNED_KEY_STATE_CURRENT_USE_RULE_ID`] (ADR 0023
    ///    current-use revalidation: historical-only or revoked signing
    ///    keys vote `Reject`) and
    ///    [`APPEND_SIGNED_TRUST_TIER_MINIMUM_RULE_ID`] (ADR 0019: the
    ///    principal MUST sit at or above `Verified`). Callers that skipped
    ///    composition are refused.
    /// 2. The key-state contributor MUST vote [`PolicyOutcome::Allow`]
    ///    regardless of final outcome. ADR 0026 §4 forbids `BreakGlass`
    ///    from substituting for current-use revalidation at the signed
    ///    ledger root.
    /// 3. A final outcome of [`PolicyOutcome::Reject`] or
    ///    [`PolicyOutcome::Quarantine`] fails closed and writes nothing.
    ///
    /// Returns the new head hash on success.
    pub fn append_signed(
        &mut self,
        mut event: Event,
        attestor: &dyn Attestor,
        policy: &PolicyDecision,
    ) -> Result<String, JsonlError> {
        require_append_signed_contributors(policy)?;
        require_append_signed_key_state_not_break_glassed(policy)?;
        require_append_final_outcome(policy, "ledger.append_signed")?;

        // Seal hash chain first so payload_hash is available for the
        // attestation preimage.
        event.prev_event_hash = self.head.clone();
        seal(&mut event);

        let signed_at = Utc::now();
        let key_id = attestor.key_id().to_string();
        let ledger_id = ledger_id_for(&self.path);
        let preimage = row_preimage(
            &event,
            &self.last_sig_prefix,
            &ledger_id,
            &key_id,
            signed_at,
        );
        let signing_input = canonical_signing_input(&preimage);
        let sig = attestor.sign(&signing_input);
        let sig_bytes = sig.to_bytes();

        let row = SignedRow {
            event,
            signature: Some(RowSignature {
                schema_version: cortex_core::canonical::SCHEMA_VERSION_ATTESTATION,
                key_id,
                signed_at,
                bytes: b64_encode(&sig_bytes),
            }),
        };
        let line = serde_json::to_string(&row).map_err(JsonlError::Encode)?;
        self.write_line(&line)?;

        self.head = Some(row.event.event_hash.clone());
        self.len += 1;
        // Advance the chain coupling.
        self.last_sig_prefix.copy_from_slice(&sig_bytes[..32]);
        Ok(row.event.event_hash)
    }

    /// Append the schema v1 -> v2 boundary event after the current v1 head,
    /// gated through the ADR 0026 enforcement lattice.
    ///
    /// This helper is intentionally narrow: it validates the typed payload and
    /// refuses to append unless the log's current head matches
    /// `payload.previous_v1_head_hash`. It does not run the full migration or
    /// bump `cortex_core::SCHEMA_VERSION`.
    ///
    /// `policy` is the composed [`PolicyDecision`] for the v1 -> v2 boundary
    /// append (ADR 0026 punch list #17). See
    /// [`Self::append_schema_migration_v1_to_v2_with_event`] for the
    /// preflight contract.
    pub fn append_schema_migration_v1_to_v2(
        &mut self,
        payload: SchemaMigrationV1ToV2Payload,
        policy: &PolicyDecision,
    ) -> Result<String, JsonlError> {
        let (head, _event) = self.append_schema_migration_v1_to_v2_with_event(payload, policy)?;
        Ok(head)
    }

    /// Schema v1 -> v2 boundary append variant that also returns the sealed
    /// [`Event`] written to the log, gated through the ADR 0026 enforcement
    /// lattice.
    ///
    /// Callers that mirror the boundary row into a side store (e.g. the
    /// SQLite events table during `cortex migrate v2` cutover) need the
    /// sealed event in hand. The JSONL row carries the canonical hash chain;
    /// the returned event is byte-identical to what was just persisted (its
    /// `prev_event_hash` and `event_hash` reflect the head couple).
    ///
    /// `policy` is the composed [`PolicyDecision`] for the v1 -> v2 boundary
    /// append and MUST satisfy:
    ///
    /// 1. The composition includes contributors for
    ///    [`SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID`] (ADR 0019 §3: the
    ///    proposing principal sits in the `Operator` authority class — a
    ///    non-operator key cannot mint a v2 boundary),
    ///    [`SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID`] (ADR 0010 §1-§2:
    ///    a fresh operator attestation is supplied over the boundary
    ///    payload), and
    ///    [`SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID`]
    ///    (ADR 0023 §2 / §5: the signing key state at attestation time is
    ///    `Active`, not `Retired` or `Revoked`). Callers that skipped
    ///    composition are refused.
    /// 2. Per ADR 0026 §4 (hard wall), the attestation contributor and the
    ///    current-use temporal-authority contributor MUST each vote
    ///    [`PolicyOutcome::Allow`] regardless of final outcome. A
    ///    `BreakGlass` decision MUST NOT substitute for either contributor at
    ///    the migration authority root.
    /// 3. A final outcome of [`PolicyOutcome::Reject`] or
    ///    [`PolicyOutcome::Quarantine`] fails closed and writes nothing.
    pub fn append_schema_migration_v1_to_v2_with_event(
        &mut self,
        payload: SchemaMigrationV1ToV2Payload,
        policy: &PolicyDecision,
    ) -> Result<(String, Event), JsonlError> {
        require_schema_migration_contributors(policy)?;
        require_schema_migration_attestation_not_break_glassed(policy)?;
        require_schema_migration_current_use_not_break_glassed(policy)?;
        require_append_final_outcome(policy, "ledger.schema_migration")?;

        let expected_head = payload.previous_v1_head_hash.clone();
        match self.head.as_deref() {
            Some(head) if head == expected_head => {}
            observed => {
                return Err(JsonlError::Validation(format!(
                    "schema migration boundary previous_v1_head_hash mismatch: observed {observed:?}, expected {expected_head}"
                )));
            }
        }

        let now = Utc::now();
        let event = schema_migration_v1_to_v2_event(payload, now, now, None)
            .map_err(|err| JsonlError::Validation(err.to_string()))?;
        // Mirror what `append` does internally so we can hand the caller the
        // identical sealed event the JSONL row carries (the post-migrate row
        // count refusal helper needs the typed Event, not just the hash, to
        // mirror the boundary into SQLite via `mirror_single_event_into_sqlite`).
        self.append_unchecked_returning_event(event)
    }

    /// Internal: append one event without the ADR 0026 policy preflight,
    /// returning both the sealed event hash and the typed Event.
    ///
    /// Used internally by
    /// [`Self::append_schema_migration_v1_to_v2_with_event`] after its own
    /// migration-specific contributors have been verified. The caller-facing
    /// Event return shape lets the schema v2 cutover mirror the boundary
    /// row into SQLite (post-migrate row-count refusal helper) without
    /// re-iterating the JSONL log. Not exposed outside the crate so
    /// external callers cannot bypass the gate on [`Self::append`].
    fn append_unchecked_returning_event(
        &mut self,
        mut event: Event,
    ) -> Result<(String, Event), JsonlError> {
        event.prev_event_hash = self.head.clone();
        seal(&mut event);
        let row = SignedRow::unsigned(event.clone());
        let line = serde_json::to_string(&row).map_err(JsonlError::Encode)?;
        self.write_line(&line)?;
        self.head = Some(event.event_hash.clone());
        self.len += 1;
        Ok((event.event_hash.clone(), event))
    }

    /// Internal: append one already-formatted JSON line + `\n` and fsync.
    fn write_line(&self, line: &str) -> Result<(), JsonlError> {
        let mut f = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)
            .map_err(|source| JsonlError::Io {
                path: self.path.clone(),
                source,
            })?;
        f.write_all(line.as_bytes())
            .map_err(|source| JsonlError::Io {
                path: self.path.clone(),
                source,
            })?;
        f.write_all(b"\n").map_err(|source| JsonlError::Io {
            path: self.path.clone(),
            source,
        })?;
        // Fsync per write — see module docs.
        f.sync_all().map_err(|source| JsonlError::Io {
            path: self.path.clone(),
            source,
        })?;
        Ok(())
    }

    /// 32-byte prefix of the most recently appended row's signature, or
    /// the genesis sentinel if no signed row has been appended yet.
    /// Exposed so callers (audit verifier, tests) can reproduce the
    /// chain coupling.
    #[must_use]
    pub fn last_sig_prefix(&self) -> [u8; 32] {
        self.last_sig_prefix
    }

    /// Iterate every row in the log in append order.
    ///
    /// Each item is the parsed [`Event`] (the inner semantic event of the
    /// signed envelope). For access to the row signature use
    /// [`Self::iter_signed`]. Decode errors short-circuit the iterator
    /// and surface as `Err`.
    pub fn iter(&self) -> Result<JsonlIter, JsonlError> {
        let mut f = File::open(&self.path).map_err(|source| JsonlError::Io {
            path: self.path.clone(),
            source,
        })?;
        f.seek(SeekFrom::Start(0))
            .map_err(|source| JsonlError::Io {
                path: self.path.clone(),
                source,
            })?;
        Ok(JsonlIter {
            path: self.path.clone(),
            reader: BufReader::new(Box::new(f) as Box<dyn Read>),
            line: 0,
        })
    }

    /// Iterate every row in the log as [`SignedRow`] envelopes — i.e.
    /// the [`Event`] plus its optional [`RowSignature`].
    ///
    /// Used by [`crate::audit::verify_signed_chain`] to reconstruct the
    /// per-row attestation preimage. Decode errors surface as `Err` and
    /// short-circuit.
    pub fn iter_signed(&self) -> Result<SignedJsonlIter, JsonlError> {
        let f = File::open(&self.path).map_err(|source| JsonlError::Io {
            path: self.path.clone(),
            source,
        })?;
        Ok(SignedJsonlIter {
            path: self.path.clone(),
            reader: BufReader::new(Box::new(f) as Box<dyn Read>),
            line: 0,
        })
    }

    /// Verify the chain end-to-end.
    ///
    /// Walks every row, recomputes both `payload_hash` and `event_hash`
    /// under the canonical framing, and confirms each row's
    /// `prev_event_hash` matches the previous row's `event_hash`. Returns
    /// `Ok(())` on success or [`JsonlError::ChainBroken`] with a short
    /// summary on failure. For per-row diagnostics use
    /// [`crate::audit::verify_chain`].
    pub fn verify_chain(&self) -> Result<(), JsonlError> {
        let mut prev: Option<String> = None;
        for (i, item) in self.iter()?.enumerate() {
            let e = item?;
            // Recompute payload_hash.
            let expected_payload = crate::hash::payload_hash(&e.payload);
            if e.payload_hash != expected_payload {
                return Err(JsonlError::ChainBroken(format!(
                    "row {} payload_hash mismatch",
                    i + 1
                )));
            }
            let expected_event =
                crate::hash::event_hash(e.prev_event_hash.as_deref(), &e.payload_hash);
            if e.event_hash != expected_event {
                return Err(JsonlError::ChainBroken(format!(
                    "row {} event_hash mismatch",
                    i + 1
                )));
            }
            if e.prev_event_hash != prev {
                return Err(JsonlError::ChainBroken(format!(
                    "row {} prev_event_hash does not point at previous row",
                    i + 1
                )));
            }
            prev = Some(e.event_hash.clone());
        }
        Ok(())
    }
}

fn require_append_final_outcome(policy: &PolicyDecision, surface: &str) -> Result<(), JsonlError> {
    match policy.final_outcome {
        PolicyOutcome::Allow | PolicyOutcome::Warn | PolicyOutcome::BreakGlass => Ok(()),
        PolicyOutcome::Quarantine | PolicyOutcome::Reject => Err(JsonlError::Validation(format!(
            "{surface} preflight: composed policy outcome {:?} blocks ledger append",
            policy.final_outcome,
        ))),
    }
}

fn require_contributor(policy: &PolicyDecision, rule_id: &str) -> Result<(), JsonlError> {
    let contains_rule = policy
        .contributing
        .iter()
        .chain(policy.discarded.iter())
        .any(|contribution| contribution.rule_id.as_str() == rule_id);
    if contains_rule {
        Ok(())
    } else {
        Err(JsonlError::Validation(format!(
            "policy decision missing required contributor `{rule_id}`; caller skipped ADR 0026 composition",
        )))
    }
}

fn require_append_contributors(policy: &PolicyDecision) -> Result<(), JsonlError> {
    require_contributor(policy, APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID)?;
    require_contributor(policy, APPEND_ATTESTATION_REQUIRED_RULE_ID)?;
    require_contributor(policy, APPEND_RUNTIME_MODE_RULE_ID)?;
    Ok(())
}

fn require_append_signed_contributors(policy: &PolicyDecision) -> Result<(), JsonlError> {
    require_contributor(policy, APPEND_SIGNED_KEY_STATE_CURRENT_USE_RULE_ID)?;
    require_contributor(policy, APPEND_SIGNED_TRUST_TIER_MINIMUM_RULE_ID)?;
    Ok(())
}

fn require_event_source_attestation(
    policy: &PolicyDecision,
    source: &EventSource,
) -> Result<(), JsonlError> {
    // ADR 0026 §4: at the User-event authority boundary, the attestation
    // contributor must itself have voted `Allow`. A BreakGlass final
    // outcome MUST NOT substitute for missing attestation on a user-sourced
    // row.
    if !matches!(source, EventSource::User) {
        return Ok(());
    }
    let attestation = policy
        .contributing
        .iter()
        .chain(policy.discarded.iter())
        .find(|contribution| {
            contribution.rule_id.as_str() == APPEND_ATTESTATION_REQUIRED_RULE_ID
        })
        .ok_or_else(|| {
            JsonlError::Validation(format!(
                "ledger.append preflight: required attestation contributor `{APPEND_ATTESTATION_REQUIRED_RULE_ID}` is absent from the policy decision for EventSource::User"
            ))
        })?;
    if attestation.outcome == PolicyOutcome::Allow {
        Ok(())
    } else {
        Err(JsonlError::Validation(format!(
            "ledger.append preflight: attestation contributor `{APPEND_ATTESTATION_REQUIRED_RULE_ID}` returned {:?} for EventSource::User; ADR 0026 §4 forbids BreakGlass substituting for attestation",
            attestation.outcome,
        )))
    }
}

fn require_schema_migration_contributors(policy: &PolicyDecision) -> Result<(), JsonlError> {
    require_contributor(policy, SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID)?;
    require_contributor(policy, SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID)?;
    require_contributor(
        policy,
        SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID,
    )?;
    Ok(())
}

fn require_schema_migration_attestation_not_break_glassed(
    policy: &PolicyDecision,
) -> Result<(), JsonlError> {
    // ADR 0026 §4: at the migration authority root, the attestation
    // contributor (ADR 0010) MUST itself have voted `Allow`. A `BreakGlass`
    // final outcome MUST NOT substitute for missing operator attestation at
    // the v1 -> v2 boundary; this is the long-promised `--unattended-migrate`
    // blocker.
    let attestation = policy
        .contributing
        .iter()
        .chain(policy.discarded.iter())
        .find(|contribution| {
            contribution.rule_id.as_str() == SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID
        })
        .ok_or_else(|| {
            JsonlError::Validation(format!(
                "ledger.schema_migration preflight: required attestation contributor `{SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID}` is absent from the policy decision"
            ))
        })?;
    if attestation.outcome == PolicyOutcome::Allow {
        Ok(())
    } else {
        Err(JsonlError::Validation(format!(
            "ledger.schema_migration preflight: attestation contributor `{SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID}` returned {:?}; ADR 0026 §4 forbids BreakGlass substituting for operator attestation at the migration authority root",
            attestation.outcome,
        )))
    }
}

fn require_schema_migration_current_use_not_break_glassed(
    policy: &PolicyDecision,
) -> Result<(), JsonlError> {
    // ADR 0023 §5 + ADR 0026 §4: the current-use temporal-authority
    // contributor MUST itself have voted `Allow`. A historical-only or
    // revoked signing key vote of `Reject` here MUST NOT be overridden by a
    // `BreakGlass` final outcome at the migration authority root.
    let current_use = policy
        .contributing
        .iter()
        .chain(policy.discarded.iter())
        .find(|contribution| {
            contribution.rule_id.as_str()
                == SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID
        })
        .ok_or_else(|| {
            JsonlError::Validation(format!(
                "ledger.schema_migration preflight: required current-use contributor `{SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID}` is absent from the policy decision"
            ))
        })?;
    if current_use.outcome == PolicyOutcome::Allow {
        Ok(())
    } else {
        Err(JsonlError::Validation(format!(
            "ledger.schema_migration preflight: current-use contributor `{SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID}` returned {:?}; ADR 0023 forbids historical-only or revoked signing keys at the migration authority root",
            current_use.outcome,
        )))
    }
}

fn require_append_signed_key_state_not_break_glassed(
    policy: &PolicyDecision,
) -> Result<(), JsonlError> {
    // ADR 0023 + ADR 0026 §4: the current-use revalidation contributor
    // must itself have voted `Allow` for a signed append. A revoked or
    // historical-only key vote of `Reject` here MUST NOT be overridden by
    // a BreakGlass final outcome on the signed ledger root.
    let key_state = policy
        .contributing
        .iter()
        .chain(policy.discarded.iter())
        .find(|contribution| {
            contribution.rule_id.as_str() == APPEND_SIGNED_KEY_STATE_CURRENT_USE_RULE_ID
        })
        .ok_or_else(|| {
            JsonlError::Validation(format!(
                "ledger.append_signed preflight: required current-use contributor `{APPEND_SIGNED_KEY_STATE_CURRENT_USE_RULE_ID}` is absent from the policy decision"
            ))
        })?;
    if key_state.outcome == PolicyOutcome::Allow {
        Ok(())
    } else {
        Err(JsonlError::Validation(format!(
            "ledger.append_signed preflight: current-use contributor `{APPEND_SIGNED_KEY_STATE_CURRENT_USE_RULE_ID}` returned {:?}; ADR 0023 forbids historical-only or revoked signing keys at the trusted ledger root",
            key_state.outcome,
        )))
    }
}

/// Build a [`PolicyDecision`] that satisfies [`JsonlLog::append`] inputs
/// for the happy path. Intended for tests and fixtures only.
///
/// Production callers MUST compose
/// [`APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID`],
/// [`APPEND_ATTESTATION_REQUIRED_RULE_ID`], and
/// [`APPEND_RUNTIME_MODE_RULE_ID`] from real trust-tier evidence, real
/// attestation evidence, and the active runtime mode. This helper is
/// exposed unconditionally because integration test crates outside
/// `cortex-ledger` need the same fixture shape; the `_test_allow` suffix
/// is the contract that documents intent.
#[must_use]
pub fn append_policy_decision_test_allow() -> PolicyDecision {
    use cortex_core::{compose_policy_outcomes, PolicyContribution};
    compose_policy_outcomes(
        vec![
            PolicyContribution::new(
                APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: event source tier gate satisfied",
            )
            .expect("static test contribution is valid"),
            PolicyContribution::new(
                APPEND_ATTESTATION_REQUIRED_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: attestation requirement satisfied",
            )
            .expect("static test contribution is valid"),
            PolicyContribution::new(
                APPEND_RUNTIME_MODE_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: runtime mode permits unsigned append",
            )
            .expect("static test contribution is valid"),
        ],
        None,
    )
}

/// Build a [`PolicyDecision`] that satisfies
/// [`JsonlLog::append_schema_migration_v1_to_v2`] inputs for the happy path.
/// Intended for tests and fixtures only.
///
/// Production callers MUST compose
/// [`SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID`] from a real ADR 0019
/// authority-class evaluation, [`SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID`]
/// from a real ADR 0010 attestation envelope verification, and
/// [`SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID`] from real
/// ADR 0023 key-state-at-event-time evidence. The `_test_allow` suffix is the
/// contract that documents intent; this helper exists so integration tests
/// outside `cortex-ledger` can build a fixture decision without re-encoding
/// the rule-id list.
#[must_use]
pub fn schema_migration_v1_to_v2_policy_decision_test_allow() -> PolicyDecision {
    use cortex_core::{compose_policy_outcomes, PolicyContribution};
    compose_policy_outcomes(
        vec![
            PolicyContribution::new(
                SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: operator authority class satisfied",
            )
            .expect("static test contribution is valid"),
            PolicyContribution::new(
                SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: operator attestation present",
            )
            .expect("static test contribution is valid"),
            PolicyContribution::new(
                SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: signing key state is current-use",
            )
            .expect("static test contribution is valid"),
        ],
        None,
    )
}

/// Build a [`PolicyDecision`] that satisfies [`JsonlLog::append_signed`]
/// inputs for the happy path. Intended for tests and fixtures only; see
/// [`append_policy_decision_test_allow`] for the production-caller
/// contract.
#[must_use]
pub fn append_signed_policy_decision_test_allow() -> PolicyDecision {
    use cortex_core::{compose_policy_outcomes, PolicyContribution};
    compose_policy_outcomes(
        vec![
            PolicyContribution::new(
                APPEND_SIGNED_KEY_STATE_CURRENT_USE_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: signing key state is current-use",
            )
            .expect("static test contribution is valid"),
            PolicyContribution::new(
                APPEND_SIGNED_TRUST_TIER_MINIMUM_RULE_ID,
                PolicyOutcome::Allow,
                "test fixture: signing principal trust tier satisfies minimum",
            )
            .expect("static test contribution is valid"),
        ],
        None,
    )
}

/// Owning iterator over a [`JsonlLog`].
pub struct JsonlIter {
    path: PathBuf,
    reader: BufReader<Box<dyn Read>>,
    line: usize,
}

impl std::fmt::Debug for JsonlIter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("JsonlIter")
            .field("path", &self.path)
            .field("line", &self.line)
            .finish()
    }
}

impl Iterator for JsonlIter {
    type Item = Result<Event, JsonlError>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            let mut buf = String::new();
            let n = match self.reader.read_line(&mut buf) {
                Ok(n) => n,
                Err(source) => {
                    return Some(Err(JsonlError::Io {
                        path: self.path.clone(),
                        source,
                    }));
                }
            };
            if n == 0 {
                return None; // EOF
            }
            self.line += 1;
            let trimmed = buf.trim();
            if trimmed.is_empty() {
                continue;
            }
            return Some(
                serde_json::from_str::<SignedRow>(trimmed)
                    .map(|row| row.event)
                    .map_err(|source| JsonlError::Decode {
                        path: self.path.clone(),
                        line: self.line,
                        source,
                    }),
            );
        }
    }
}

/// Owning iterator over a [`JsonlLog`] yielding full [`SignedRow`]
/// envelopes (event + optional signature). Used by the Ed25519-aware
/// audit verifier in [`crate::audit::verify_signed_chain`].
pub struct SignedJsonlIter {
    path: PathBuf,
    reader: BufReader<Box<dyn Read>>,
    line: usize,
}

impl std::fmt::Debug for SignedJsonlIter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SignedJsonlIter")
            .field("path", &self.path)
            .field("line", &self.line)
            .finish()
    }
}

impl Iterator for SignedJsonlIter {
    type Item = Result<SignedRow, JsonlError>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            let mut buf = String::new();
            let n = match self.reader.read_line(&mut buf) {
                Ok(n) => n,
                Err(source) => {
                    return Some(Err(JsonlError::Io {
                        path: self.path.clone(),
                        source,
                    }));
                }
            };
            if n == 0 {
                return None;
            }
            self.line += 1;
            let trimmed = buf.trim();
            if trimmed.is_empty() {
                continue;
            }
            return Some(
                serde_json::from_str::<SignedRow>(trimmed).map_err(|source| JsonlError::Decode {
                    path: self.path.clone(),
                    line: self.line,
                    source,
                }),
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::TimeZone;
    use cortex_core::{
        compose_policy_outcomes, Event, EventId, EventSource, EventType, PolicyContribution,
        SchemaMigrationV1ToV2Payload, SCHEMA_MIGRATION_V1_TO_V2_EVENT_KIND,
        SCHEMA_MIGRATION_V1_TO_V2_TARGET, SCHEMA_VERSION,
    };
    use tempfile::tempdir;

    fn fixture_event(seq: u64) -> Event {
        Event {
            id: EventId::new(),
            schema_version: SCHEMA_VERSION,
            observed_at: chrono::Utc.with_ymd_and_hms(2026, 1, 1, 12, 0, 0).unwrap(),
            recorded_at: chrono::Utc.with_ymd_and_hms(2026, 1, 1, 12, 0, 1).unwrap(),
            source: EventSource::User,
            event_type: EventType::UserMessage,
            trace_id: None,
            session_id: None,
            domain_tags: vec![],
            payload: serde_json::json!({"seq": seq, "text": format!("event {seq}")}),
            payload_hash: String::new(),
            prev_event_hash: None,
            event_hash: String::new(),
        }
    }

    fn allow_policy() -> PolicyDecision {
        append_policy_decision_test_allow()
    }

    fn migration_allow_policy() -> PolicyDecision {
        schema_migration_v1_to_v2_policy_decision_test_allow()
    }

    /// T-1.B.2 acceptance: append N events, reopen, verify chain.
    #[test]
    fn append_n_reopen_and_verify() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("events.jsonl");

        // Append 25 events.
        let mut log = JsonlLog::open(&path).unwrap();
        let mut heads = Vec::new();
        let policy = allow_policy();
        for i in 0..25u64 {
            let head = log.append(fixture_event(i), &policy).unwrap();
            heads.push(head);
        }
        assert_eq!(log.len(), 25);
        assert_eq!(log.head(), Some(heads.last().unwrap().as_str()));

        // Reopen and verify.
        let log2 = JsonlLog::open(&path).unwrap();
        assert_eq!(log2.len(), 25);
        assert_eq!(log2.head(), Some(heads.last().unwrap().as_str()));
        log2.verify_chain().expect("chain verifies after reopen");

        // Iterate and confirm prev->next linkage.
        let mut prev: Option<String> = None;
        let mut count = 0;
        for item in log2.iter().unwrap() {
            let e = item.unwrap();
            assert_eq!(e.prev_event_hash, prev);
            prev = Some(e.event_hash.clone());
            count += 1;
        }
        assert_eq!(count, 25);
    }

    #[test]
    fn empty_log_verifies() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("empty.jsonl");
        let log = JsonlLog::open(&path).unwrap();
        assert_eq!(log.len(), 0);
        assert!(log.head().is_none());
        log.verify_chain().expect("empty chain is valid");
    }

    #[test]
    fn append_persists_after_drop() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("persist.jsonl");
        {
            let mut log = JsonlLog::open(&path).unwrap();
            let policy = allow_policy();
            log.append(fixture_event(0), &policy).unwrap();
            log.append(fixture_event(1), &policy).unwrap();
            // drop log; fsync should have made the rows durable.
        }
        let log2 = JsonlLog::open(&path).unwrap();
        assert_eq!(log2.len(), 2);
    }

    #[test]
    fn corrupted_payload_fails_verify() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("corrupt.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        let policy = allow_policy();
        log.append(fixture_event(0), &policy).unwrap();
        log.append(fixture_event(1), &policy).unwrap();
        log.append(fixture_event(2), &policy).unwrap();

        // Tamper: rewrite the file with a mutated payload on row 2.
        let lines: Vec<String> = std::fs::read_to_string(&path)
            .unwrap()
            .lines()
            .map(|s| s.to_string())
            .collect();
        let mut bad: Event = serde_json::from_str(&lines[1]).unwrap();
        bad.payload = serde_json::json!({"tampered": true});
        let mut new_content = String::new();
        new_content.push_str(&lines[0]);
        new_content.push('\n');
        new_content.push_str(&serde_json::to_string(&bad).unwrap());
        new_content.push('\n');
        new_content.push_str(&lines[2]);
        new_content.push('\n');
        std::fs::write(&path, new_content).unwrap();

        let log2 = JsonlLog::open(&path).unwrap();
        let err = log2.verify_chain().unwrap_err();
        assert!(matches!(err, JsonlError::ChainBroken(_)));
    }

    #[test]
    fn append_after_reopen_continues_chain() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("continue.jsonl");
        let head_before;
        {
            let mut log = JsonlLog::open(&path).unwrap();
            let policy = allow_policy();
            log.append(fixture_event(0), &policy).unwrap();
            head_before = log.append(fixture_event(1), &policy).unwrap();
        }
        let mut log2 = JsonlLog::open(&path).unwrap();
        assert_eq!(log2.head(), Some(head_before.as_str()));
        let head_after = log2.append(fixture_event(2), &allow_policy()).unwrap();
        assert_ne!(head_after, head_before);
        log2.verify_chain().expect("continued chain verifies");
    }

    #[test]
    fn schema_migration_v1_to_v2_event_emitted_after_v1_head() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("schema-boundary.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        let previous_v1_head = log.append(fixture_event(0), &allow_policy()).unwrap();
        let payload = SchemaMigrationV1ToV2Payload::new(
            previous_v1_head.clone(),
            "script-digest",
            None,
            "fixture-digest",
        );

        let boundary_head = log
            .append_schema_migration_v1_to_v2(payload, &migration_allow_policy())
            .expect("boundary event appends");

        assert_eq!(log.len(), 2);
        assert_eq!(log.head(), Some(boundary_head.as_str()));
        log.verify_chain().expect("boundary chain verifies");

        let rows = log.iter().unwrap().collect::<Result<Vec<_>, _>>().unwrap();
        let boundary = rows.last().expect("boundary row exists");
        assert_eq!(boundary.schema_version, SCHEMA_MIGRATION_V1_TO_V2_TARGET);
        assert_eq!(boundary.event_type, EventType::SystemNote);
        assert_eq!(boundary.source, EventSource::Runtime);
        assert_eq!(
            boundary.prev_event_hash.as_deref(),
            Some(previous_v1_head.as_str())
        );
        assert_eq!(
            boundary.payload["kind"],
            SCHEMA_MIGRATION_V1_TO_V2_EVENT_KIND
        );
        assert_eq!(
            boundary.payload["payload"]["previous_v1_head_hash"],
            previous_v1_head
        );
    }

    #[test]
    fn schema_migration_v1_to_v2_event_rejects_wrong_previous_head() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("schema-boundary-mismatch.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        log.append(fixture_event(0), &allow_policy()).unwrap();
        let payload =
            SchemaMigrationV1ToV2Payload::new("not-current-head", "script-digest", None, "fixture");

        let err = log
            .append_schema_migration_v1_to_v2(payload, &migration_allow_policy())
            .expect_err("wrong previous head must fail");

        assert!(matches!(err, JsonlError::Validation(_)));
        assert_eq!(log.len(), 1);
        log.verify_chain()
            .expect("rejected boundary append must not corrupt chain");
    }

    #[test]
    fn schema_migration_v1_to_v2_refuses_missing_authority_class_contributor() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("schema-boundary-missing-auth.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        let previous_v1_head = log.append(fixture_event(0), &allow_policy()).unwrap();
        let payload = SchemaMigrationV1ToV2Payload::new(
            previous_v1_head,
            "script-digest",
            None,
            "fixture-digest",
        );

        // Policy is missing the authority-class contributor.
        let policy = compose_policy_outcomes(
            vec![
                PolicyContribution::new(
                    SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: attestation present",
                )
                .unwrap(),
                PolicyContribution::new(
                    SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: current use",
                )
                .unwrap(),
            ],
            None,
        );

        let err = log
            .append_schema_migration_v1_to_v2(payload, &policy)
            .expect_err("missing authority-class contributor must fail");
        assert!(matches!(err, JsonlError::Validation(_)));
        assert_eq!(log.len(), 1);
        log.verify_chain()
            .expect("rejected boundary append must not corrupt chain");
    }

    #[test]
    fn schema_migration_v1_to_v2_refuses_reject_outcome() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("schema-boundary-reject.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        let previous_v1_head = log.append(fixture_event(0), &allow_policy()).unwrap();
        let payload = SchemaMigrationV1ToV2Payload::new(
            previous_v1_head,
            "script-digest",
            None,
            "fixture-digest",
        );

        // All three contributors present; authority-class refuses with Reject.
        let policy = compose_policy_outcomes(
            vec![
                PolicyContribution::new(
                    SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID,
                    PolicyOutcome::Reject,
                    "fixture: non-operator authority class",
                )
                .unwrap(),
                PolicyContribution::new(
                    SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: attestation present",
                )
                .unwrap(),
                PolicyContribution::new(
                    SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: current use",
                )
                .unwrap(),
            ],
            None,
        );
        assert_eq!(policy.final_outcome, PolicyOutcome::Reject);

        let err = log
            .append_schema_migration_v1_to_v2(payload, &policy)
            .expect_err("reject outcome must fail closed");
        assert!(matches!(err, JsonlError::Validation(_)));
        assert_eq!(log.len(), 1);
    }

    #[test]
    fn schema_migration_v1_to_v2_refuses_quarantine_outcome() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("schema-boundary-quarantine.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        let previous_v1_head = log.append(fixture_event(0), &allow_policy()).unwrap();
        let payload = SchemaMigrationV1ToV2Payload::new(
            previous_v1_head,
            "script-digest",
            None,
            "fixture-digest",
        );

        let policy = compose_policy_outcomes(
            vec![
                PolicyContribution::new(
                    SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID,
                    PolicyOutcome::Quarantine,
                    "fixture: under-trust authority class",
                )
                .unwrap(),
                PolicyContribution::new(
                    SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: attestation present",
                )
                .unwrap(),
                PolicyContribution::new(
                    SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: current use",
                )
                .unwrap(),
            ],
            None,
        );
        assert_eq!(policy.final_outcome, PolicyOutcome::Quarantine);

        let err = log
            .append_schema_migration_v1_to_v2(payload, &policy)
            .expect_err("quarantine outcome must fail closed");
        assert!(matches!(err, JsonlError::Validation(_)));
        assert_eq!(log.len(), 1);
    }

    #[test]
    fn schema_migration_v1_to_v2_refuses_attestation_break_glass() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("schema-boundary-break-glass.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        let previous_v1_head = log.append(fixture_event(0), &allow_policy()).unwrap();
        let payload = SchemaMigrationV1ToV2Payload::new(
            previous_v1_head,
            "script-digest",
            None,
            "fixture-digest",
        );

        // Attestation contributor voted BreakGlass; ADR 0026 §4 forbids
        // BreakGlass from substituting for operator attestation here.
        let policy = compose_policy_outcomes(
            vec![
                PolicyContribution::new(
                    SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: operator authority class",
                )
                .unwrap(),
                PolicyContribution::new(
                    SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID,
                    PolicyOutcome::BreakGlass,
                    "fixture: operator attempted break-glass on attestation",
                )
                .unwrap(),
                PolicyContribution::new(
                    SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: current use",
                )
                .unwrap(),
            ],
            None,
        );

        let err = log
            .append_schema_migration_v1_to_v2(payload, &policy)
            .expect_err("BreakGlass on attestation must fail");
        assert!(matches!(err, JsonlError::Validation(_)));
        assert_eq!(log.len(), 1);
    }

    #[test]
    fn schema_migration_v1_to_v2_refuses_historical_key_current_use() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("schema-boundary-historical-key.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        let previous_v1_head = log.append(fixture_event(0), &allow_policy()).unwrap();
        let payload = SchemaMigrationV1ToV2Payload::new(
            previous_v1_head,
            "script-digest",
            None,
            "fixture-digest",
        );

        // Current-use contributor refuses because the signing key is
        // historical-only (Retired or Revoked at attestation time).
        let policy = compose_policy_outcomes(
            vec![
                PolicyContribution::new(
                    SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: operator authority class",
                )
                .unwrap(),
                PolicyContribution::new(
                    SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: attestation present",
                )
                .unwrap(),
                PolicyContribution::new(
                    SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID,
                    PolicyOutcome::Reject,
                    "fixture: signing key retired before attestation time",
                )
                .unwrap(),
            ],
            None,
        );

        let err = log
            .append_schema_migration_v1_to_v2(payload, &policy)
            .expect_err("historical-only signing key must fail");
        assert!(matches!(err, JsonlError::Validation(_)));
        assert_eq!(log.len(), 1);
    }

    #[test]
    fn append_refuses_policy_decision_missing_contributors() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("missing-contributor.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        // Compose a policy with only one of the three required contributors.
        let policy = compose_policy_outcomes(
            vec![PolicyContribution::new(
                APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID,
                PolicyOutcome::Allow,
                "fixture: tier gate only",
            )
            .unwrap()],
            None,
        );

        let err = log
            .append(fixture_event(0), &policy)
            .expect_err("missing contributor must fail");
        assert!(matches!(err, JsonlError::Validation(_)));
        assert_eq!(log.len(), 0);
    }

    #[test]
    fn append_refuses_reject_outcome() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("reject-outcome.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        let policy = compose_policy_outcomes(
            vec![
                PolicyContribution::new(
                    APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID,
                    PolicyOutcome::Reject,
                    "fixture: tier gate refuses",
                )
                .unwrap(),
                PolicyContribution::new(
                    APPEND_ATTESTATION_REQUIRED_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: attestation present",
                )
                .unwrap(),
                PolicyContribution::new(
                    APPEND_RUNTIME_MODE_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: runtime mode permits unsigned",
                )
                .unwrap(),
            ],
            None,
        );
        assert_eq!(policy.final_outcome, PolicyOutcome::Reject);

        let err = log
            .append(fixture_event(0), &policy)
            .expect_err("reject outcome must fail closed");
        assert!(matches!(err, JsonlError::Validation(_)));
        assert_eq!(log.len(), 0);
    }

    #[test]
    fn append_refuses_user_event_when_attestation_contributor_not_allow() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("user-attestation.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        // Attestation says Warn, not Allow; ADR 0026 §4 forbids BreakGlass /
        // Warn substituting for attestation at the User authority boundary.
        let policy = compose_policy_outcomes(
            vec![
                PolicyContribution::new(
                    APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: tier gate allows",
                )
                .unwrap(),
                PolicyContribution::new(
                    APPEND_ATTESTATION_REQUIRED_RULE_ID,
                    PolicyOutcome::Warn,
                    "fixture: attestation warning",
                )
                .unwrap(),
                PolicyContribution::new(
                    APPEND_RUNTIME_MODE_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: runtime mode permits unsigned",
                )
                .unwrap(),
            ],
            None,
        );
        // Final outcome is Warn, but EventSource::User still must fail.
        let err = log
            .append(fixture_event(0), &policy)
            .expect_err("User event without Allow attestation must fail");
        assert!(matches!(err, JsonlError::Validation(_)));
        assert_eq!(log.len(), 0);
    }

    #[test]
    fn append_allows_warn_outcome() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("warn-outcome.jsonl");
        let mut log = JsonlLog::open(&path).unwrap();
        // Warn final outcome with Allow attestation must still append for
        // an EventSource::User row (the local-development ledger pattern).
        let policy = compose_policy_outcomes(
            vec![
                PolicyContribution::new(
                    APPEND_EVENT_SOURCE_TIER_GATE_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: tier gate allows",
                )
                .unwrap(),
                PolicyContribution::new(
                    APPEND_ATTESTATION_REQUIRED_RULE_ID,
                    PolicyOutcome::Allow,
                    "fixture: attestation present",
                )
                .unwrap(),
                PolicyContribution::new(
                    APPEND_RUNTIME_MODE_RULE_ID,
                    PolicyOutcome::Warn,
                    "fixture: runtime mode is local-development",
                )
                .unwrap(),
            ],
            None,
        );
        assert_eq!(policy.final_outcome, PolicyOutcome::Warn);
        log.append(fixture_event(0), &policy)
            .expect("warn outcome must still append");
        assert_eq!(log.len(), 1);
    }
}