khive_runtime/

audit_batch.rs

//! ADR-133 Slice 1: the audit-batch seam.
//!
//! Incidental audit writes (gate denials, dispatch outcomes, config-lock
//! rows, git.digest receipts, and pure-observability rows like
//! `RecallExecuted`) no longer take one writer-task acquisition per row on
//! the request hot path. Concurrent submissions arriving while a generation
//! is committing share the *next* generation instead of each taking their
//! own writer acquisition, so N concurrent producers collapse to one
//! [`khive_storage::EventStore::append_events_idempotent`] call per
//! generation rather than N.
//!
//! [`AuditBatch`] owns this seam. A lazily-spawned supervisor task drains
//! pending rows into generations and drives each through the store; the
//! supervisor's own `JoinHandle` is retained (never discarded) so an
//! abnormal exit — panic, cancellation, a lost child join, or a driver that
//! returns `Ok` while state is not terminally consistent — is observed and
//! converted into a `Failed` transition with all accepted waiters resolved,
//! per owner ruling R1/R4 (`.khive/OWNER_RULING_adr133_gate.md`).

#[cfg(any(test, feature = "fault-injection"))]
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;

use parking_lot::Mutex;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;

use khive_storage::event::EventAppendDisposition;
use khive_storage::{Event, EventStore, StorageError, WriterTaskRequestState};

/// Coarse durability classification for an [`AuditProducer`]. See
/// [`classify`] for the exhaustive mapping.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum AuditProductionClass {
    /// The dispatch this row audits carries an obligation: the caller-visible
    /// outcome must not silently diverge from what was durably recorded.
    DispatchObligation,
    /// The row is a best-effort observability signal; a non-commit degrades
    /// gracefully rather than blocking or failing the dispatch it audits.
    PureObservability,
}

/// Every call site that can submit a row through [`AuditBatchControl`].
/// Adding a variant here without extending the crate-private `classify`
/// function's match is a compile error — there is no wildcard arm.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuditProducer {
    /// The gate denied a dispatch; the denial itself is audited.
    GateDenied,
    /// The gate backend was unreachable and the dispatch failed closed; the
    /// refusal is audited best-effort: the typed `GateUnavailable` refusal
    /// already fails the dispatch, so a lost row degrades diagnostics only.
    GateUnavailable,
    /// A pack dispatch returned a successful result.
    DispatchSucceeded,
    /// A pack dispatch returned an error result.
    DispatchFailed,
    /// A redirected KG read consulted the gate on its effective entity id.
    EffectiveTargetCheck,
    /// The gate allowed a verb no pack owns.
    UnknownVerb,
    /// The strict `git.digest` success receipt (schema v2).
    GitDigestReceipt,
    /// A drained process-lifetime `OnceLock` config-lock row.
    ConfigLocked,
    /// A `memory.recall` execution's pure-observability audit row.
    ///
    /// Classified and ready for routing, but not yet wired to a live call
    /// site in this slice: `khive-pack-memory`'s recall handler reaches only
    /// `KhiveRuntime` (`crates/khive-runtime/src/runtime.rs`), which does not
    /// hold this batch seam and is outside this change's file ownership
    /// (`final_file_ownership_r2.md` assigns `runtime.rs` to the D8 author).
    /// Wiring this variant to `emit_recall_executed_event` needs either an
    /// ownership-map amendment granting `runtime.rs` a narrow accessor, or
    /// threading a second seam onto `KhiveRuntime` — both out of scope here.
    #[allow(dead_code)]
    RecallExecuted,
}

/// Classify an [`AuditProducer`] into its [`AuditProductionClass`]. One
/// exhaustive match, no wildcard arm.
pub(crate) const fn classify(producer: AuditProducer) -> AuditProductionClass {
    match producer {
        AuditProducer::GateDenied
        | AuditProducer::DispatchSucceeded
        | AuditProducer::DispatchFailed
        | AuditProducer::EffectiveTargetCheck
        | AuditProducer::UnknownVerb
        | AuditProducer::GitDigestReceipt => AuditProductionClass::DispatchObligation,
        AuditProducer::ConfigLocked
        | AuditProducer::RecallExecuted
        | AuditProducer::GateUnavailable => AuditProductionClass::PureObservability,
    }
}

/// Exhaustive terminal reasons an [`AuditBatchControl::submit`],
/// [`AuditBatchControl::quiesce`], or [`AuditBatchControl::close_and_drain`]
/// call can resolve to. Never mapped through a wildcard arm anywhere in this
/// module (R4).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuditTerminalReason {
    /// `EventStore::preflight_event` rejected the row before it was ever
    /// enqueued. The row was never counted as submitted.
    PreflightRejected,
    /// The batch is `Closing` or `Closed`; new admission is refused.
    AdmissionClosed,
    /// `AuditBatchConfig::max_pending_rows` was reached before this row was
    /// enqueued. The row was never counted as submitted and never shared a
    /// generation with anyone — safe to retry, and doing so applies the
    /// obligation at most once.
    QueueAdmissionExhausted,
    /// The row was already enqueued (counted in `submitted_rows`) when this
    /// caller's `AuditBatchConfig::admission_deadline` elapsed waiting for
    /// its generation's outcome. Unlike [`Self::QueueAdmissionExhausted`],
    /// the row was not refused: by the moment the deadline fires it may
    /// still be sitting in `state.pending`, or the driver may have already
    /// drained it into an in-flight generation — either way it remains
    /// enqueued and unresolved, and is committed (or terminally failed) by
    /// the generation driver independently of this caller's timeout, so the
    /// caller cannot tell from this reason alone whether the row eventually
    /// committed, or even which of those two states it was in when the
    /// deadline elapsed. Retrying is only safe for an idempotent caller —
    /// the prior submission may still land. When this reason degrades an
    /// admission-degrade-safe read's own audit obligation, it is counted
    /// separately from [`Self::QueueAdmissionExhausted`] — see
    /// `pack::audit_admission_unresolved_obligation_count` — precisely
    /// because a row counted here may still commit, unlike one refused
    /// before enqueue.
    AdmissionDeadlineExpired,
    /// Reached only through `AuditBatch::submit_until_resolved`: the row's
    /// [`Self::AdmissionDeadlineExpired`] wait had already elapsed, and this
    /// caller's own `AuditBatchConfig::resolution_deadline` then also
    /// elapsed still waiting for the row's real generation outcome. The
    /// caller reaches this only after its domain effect has already
    /// committed, so the effect is never retried and the row is never
    /// re-enqueued — but unlike an ordinary commit, the caller now learns
    /// the effect committed while the audit outcome itself is unresolved.
    /// Same as [`Self::AdmissionDeadlineExpired`], the row is left exactly
    /// where the driver holds it (`state.pending`, or already mid-generation)
    /// for the driver to resolve independently — this reason performs no
    /// removal. Kept distinct from [`Self::AdmissionDeadlineExpired`] so a
    /// caller, and diagnostics reading this reason, can tell a merely-slow
    /// admission wait apart from a resolution wait that gave up entirely.
    ResolutionDeadlineExpired,
    /// A row shared this generation's id with a previously stored row whose
    /// columns or observation projection did not match exactly.
    IdentityConflict,
    /// `classify_store_error` judged the store's error non-retryable, and the
    /// generation stopped on that attempt. Usually that is the first attempt,
    /// but not necessarily: a generation whose earlier attempts failed
    /// retryably and whose next one returns a non-retryable error reports
    /// this reason too, because the attempt that decided the outcome is the
    /// non-retryable one. Distinct from [`Self::RetryExhausted`], which the
    /// classifier judged safe to retry on every attempt and which failed
    /// anyway once they ran out — this reason carries no such hope. Whatever
    /// the store returned on the deciding attempt, it is not the kind of
    /// failure `AuditBatchConfig::max_commit_attempts` exists to ride out,
    /// so a caller or an automated retry policy reading this reason should
    /// not schedule a bare retry of the same call and should instead treat
    /// it as a storage fault needing attention.
    StoreFailure,
    /// The store returned a `classify_store_error`-retryable error on every
    /// one of `AuditBatchConfig::max_commit_attempts` attempts for this
    /// generation, and the last attempt still failed retryable. Kept
    /// distinct from [`Self::StoreFailure`] — the same way
    /// [`Self::ResolutionDeadlineExpired`] is kept distinct from
    /// [`Self::AdmissionDeadlineExpired`] — so a caller, and diagnostics
    /// reading this reason, can tell a store call the classifier judged
    /// hopeless apart from one that kept failing a condition (write-queue or
    /// writer-task pressure, pool or timeout) the classifier judged
    /// transient. The underlying condition may still be
    /// transient at the moment attempts run out (a daemon restart, pool
    /// pressure outlasting the configured backoff), so an operator or an
    /// automated retry policy sitting above this batch can choose to wait
    /// longer and try again rather than treating it identically to
    /// [`Self::StoreFailure`]. This reason changes no tolerance, deadline,
    /// retry count, or backoff on its own — it only names which of the two
    /// causes produced the generation's failure.
    RetryExhausted,
    /// The configured `EventStore` backend does not implement
    /// `append_events_idempotent`.
    IdempotencyUnsupported,
    /// The generation driver task panicked, or the supervisor awaiting it
    /// unwound while armed.
    DriverPanicked,
    /// The generation driver task was cancelled/aborted, or the supervisor
    /// awaiting it was dropped mid-await (shutdown abort) while armed.
    DriverCancelled,
    /// The child driver's `JoinHandle` was lost — dropped without ever being
    /// inspected — so its outcome could not be classified.
    DriverJoinLost,
    /// The driver returned `Ok` but locked batch state was not proved
    /// terminally consistent afterward (`in_flight` still set, or the
    /// generation's rows were never resolved).
    DriverExitedInconsistent,
    /// The driver's own bound on a single generation's
    /// `EventStore::append_events_idempotent()` call
    /// (`supervisor_loop`'s `driver_append_deadline`) elapsed while that
    /// call was still in flight. Unlike [`Self::AdmissionDeadlineExpired`]
    /// and [`Self::ResolutionDeadlineExpired`], which bound only how long a
    /// *caller* keeps waiting while the row stays wherever the driver holds
    /// it, this bounds the driver's own hold: every waiter on the abandoned
    /// generation is resolved with this reason and the generation is
    /// removed from the driver's in-flight state, so a stalled store call
    /// cannot pin `pending` at `max_pending_rows` and starve all later
    /// admission (khive#2331). The underlying append is not cancelled — it
    /// may not be safely cancellable mid-flight — so it is left to run to
    /// completion in the background and its eventual result, whatever it
    /// is, is discarded; no waiter is still listening for it. The caller's
    /// domain effect (if any already committed before this row was
    /// enqueued) is never retried, and the row is never re-enqueued.
    DriverAppendAbandoned,
    /// Before spawning a generation's child, the driver found
    /// `AuditBatchConfig::max_abandoned_appends` detached appends already
    /// outstanding from prior [`Self::DriverAppendAbandoned`] generations
    /// (khive#2331). The store is treated as wedged: this generation's rows
    /// are shed without ever attempting an append — no child task is
    /// spawned, no store call is made, and every waiter resolves with this
    /// reason immediately, well inside `driver_append_deadline`. Same
    /// non-retry contract as [`Self::DriverAppendAbandoned`]: any
    /// already-committed domain effect is not retried and the row is never
    /// re-enqueued, and pure-observability producers record degradation.
    /// The driver reattempts an append on the next generation as soon as an
    /// outstanding append returns (commit or failure) and the count drops
    /// back below the cap — recovery needs no timer of its own.
    StoreWedged,
}

/// One row accepted for batching: the immutable event identity plus the
/// producer that minted it, used for classification and, on failure,
/// degradation accounting.
pub struct PreparedAuditRow {
    pub event: Event,
    pub producer: AuditProducer,
}

/// What [`AuditBatchControl::submit`] resolves to on a non-error outcome.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuditCommitOutcome {
    /// The row was freshly inserted.
    Committed,
    /// A prior row with the same identity already matched exactly (internal
    /// retry replaying the same producer-minted identity).
    AlreadyPresentIdentical,
}

/// Production-visible snapshot of [`AuditBatch::health_metrics`]. See there
/// for field semantics. `khive_db::diagnostics::RuntimeAuditBatchMetrics` carries
/// these three fields plus `admission_refused_obligations` and
/// `admission_unresolved_obligations`, which are sourced from process-wide
/// counters outside `AuditBatch` rather than from this struct — see
/// `VerbRegistry::audit_batch_metrics`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AuditBatchHealthMetrics {
    pub flush_failures: u64,
    pub degraded_rows: u64,
    pub degraded: bool,
    /// Detached [`AuditTerminalReason::DriverAppendAbandoned`] appends that
    /// eventually returned a commit, after their generation had already
    /// resolved every waiter with `DriverAppendAbandoned`. Lets an operator
    /// see that a store recorded as wedged later drained.
    pub late_append_commits: u64,
    /// Detached [`AuditTerminalReason::DriverAppendAbandoned`] appends that
    /// eventually returned a failure (store error, panic, or cancellation),
    /// after their generation had already resolved every waiter with
    /// `DriverAppendAbandoned`.
    pub late_append_failures: u64,
}

/// Tunables for the batch seam. Defaults are conservative; every field is
/// exercised by at least one mechanism test.
#[derive(Debug, Clone)]
pub struct AuditBatchConfig {
    pub max_pending_rows: std::num::NonZeroUsize,
    pub max_rows_per_generation: std::num::NonZeroUsize,
    pub max_commit_attempts: std::num::NonZeroU8,
    pub retry_backoff: Duration,
    pub admission_deadline: Duration,
    /// Caps how much longer `AuditBatch::submit_until_resolved` keeps
    /// waiting on a row's real generation outcome once `admission_deadline`
    /// has already elapsed on it. Without this bound a generation stuck on a
    /// stalled `EventStore::append_events_idempotent` call retains the
    /// completed write's caller, its request slot, and its audit-lane waiter
    /// forever, exhausting both request and audit capacity (khive#2331).
    /// Must be at least `admission_deadline` — validated (debug-only) in
    /// [`AuditBatch::new`]. Defaults to 6x `admission_deadline`.
    ///
    /// Also the single source value the driver's own per-generation append
    /// bound (`supervisor_loop`'s `driver_append_deadline`) is derived from
    /// — that bound exists so a stalled append cannot keep the driver from
    /// ever draining `pending` again, which is what actually exhausts
    /// admission for every later caller, not just the one already waiting
    /// on the stuck row. See [`AuditTerminalReason::DriverAppendAbandoned`].
    pub resolution_deadline: Duration,
    /// Caps how many [`AuditTerminalReason::DriverAppendAbandoned`] appends
    /// may be outstanding (handed to a detached task, still running) at
    /// once. Each outstanding append retains up to
    /// `max_rows_per_generation` events until it finally returns, so the
    /// retained-buffer bound this places on a wedged store is
    /// `max_abandoned_appends * max_rows_per_generation` rows — without it,
    /// a store whose append never returns mints one such task per
    /// `driver_append_deadline` with no cap, and both the live task count
    /// and the retained event batches inside them grow until the process
    /// exits (khive#2331). Once the cap is reached, `supervisor_loop` sheds
    /// further generations with [`AuditTerminalReason::StoreWedged`] instead
    /// of attempting another append; it resumes attempting appends as soon
    /// as an outstanding one returns and the count drops back below the
    /// cap. Defaults to 4.
    pub max_abandoned_appends: std::num::NonZeroUsize,
}

impl Default for AuditBatchConfig {
    fn default() -> Self {
        let admission_deadline = Duration::from_secs(5);
        Self {
            max_pending_rows: std::num::NonZeroUsize::new(4096).unwrap(),
            max_rows_per_generation: std::num::NonZeroUsize::new(256).unwrap(),
            max_commit_attempts: std::num::NonZeroU8::new(3).unwrap(),
            retry_backoff: Duration::from_millis(20),
            admission_deadline,
            resolution_deadline: admission_deadline * 6,
            max_abandoned_appends: std::num::NonZeroUsize::new(4).unwrap(),
        }
    }
}

#[async_trait::async_trait]
pub trait AuditBatchControl: Send + Sync {
    async fn submit(
        &self,
        row: PreparedAuditRow,
    ) -> Result<AuditCommitOutcome, AuditTerminalReason>;
    async fn quiesce(&self) -> Result<(), AuditTerminalReason>;
    async fn close_and_drain(&self) -> Result<(), AuditTerminalReason>;
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum Lifecycle {
    Open,
    Closing,
    Closed,
    Failed(AuditTerminalReason),
}

struct Waiting {
    event: Event,
    producer: AuditProducer,
    responder: oneshot::Sender<Result<AuditCommitOutcome, AuditTerminalReason>>,
}

/// A committed/failed generation's accounting, retained for the lifetime of
/// the process (bounded in practice by process lifetime and generation
/// volume; this slice makes no attempt to prune history).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AuditGenerationSnapshot {
    pub generation_id: u64,
    pub submitted_rows: u64,
    pub committed_rows: u64,
    pub store_batch_calls: u64,
    pub terminal_reason: Option<AuditTerminalReason>,
}

struct State {
    lifecycle: Lifecycle,
    pending: Vec<Waiting>,
    in_flight_generation: Option<u64>,
    driver_active: bool,
    next_generation_id: u64,
    submitted_rows: u64,
    committed_rows: u64,
    store_batch_calls: u64,
    generations: Vec<AuditGenerationSnapshot>,
    flush_failures: u64,
    degraded_rows: u64,
    degraded: bool,
    /// Detached `DriverAppendAbandoned` appends currently still running,
    /// bounded by `AuditBatchConfig::max_abandoned_appends`. Incremented
    /// when `supervisor_loop` abandons a generation and hands its child to a
    /// detached task; decremented by that task once the child finally
    /// returns.
    outstanding_abandoned_appends: usize,
    late_append_commits: u64,
    late_append_failures: u64,
}

impl State {
    fn new() -> Self {
        Self {
            lifecycle: Lifecycle::Open,
            pending: Vec::new(),
            in_flight_generation: None,
            driver_active: false,
            next_generation_id: 0,
            submitted_rows: 0,
            committed_rows: 0,
            store_batch_calls: 0,
            generations: Vec::new(),
            flush_failures: 0,
            degraded_rows: 0,
            degraded: false,
            outstanding_abandoned_appends: 0,
            late_append_commits: 0,
            late_append_failures: 0,
        }
    }

    fn is_idle(&self) -> bool {
        self.pending.is_empty() && self.in_flight_generation.is_none() && !self.driver_active
    }
}

struct Inner {
    state: Mutex<State>,
}

impl Inner {
    /// Wins once: the first abnormal driver/supervisor exit sets `Failed`
    /// and drains every accepted waiter (this generation's, plus anything
    /// still queued for a future one) with the same typed reason. A later
    /// abnormal path observes the existing terminal state and does not
    /// double-count (R1).
    fn fail_driver(&self, reason: AuditTerminalReason, in_flight_waiters: Vec<Waiting>) {
        let mut state = self.state.lock();
        let already_failed = matches!(state.lifecycle, Lifecycle::Failed(_));
        if !already_failed {
            state.lifecycle = Lifecycle::Failed(reason);
        }
        state.driver_active = false;
        state.in_flight_generation = None;
        let drained: Vec<Waiting> = std::mem::take(&mut state.pending);
        if !already_failed {
            state.flush_failures += 1;
            state.degraded = true;
            let generation_id = state.next_generation_id;
            state.generations.push(AuditGenerationSnapshot {
                generation_id,
                submitted_rows: (in_flight_waiters.len() + drained.len()) as u64,
                committed_rows: 0,
                store_batch_calls: 0,
                terminal_reason: Some(reason),
            });
            state.next_generation_id += 1;
        }
        drop(state);
        for waiting in in_flight_waiters.into_iter().chain(drained) {
            record_degradation_if_pure(self, waiting.producer);
            let _ = waiting.responder.send(Err(reason));
        }
    }

    fn record_degradation(&self) {
        let mut state = self.state.lock();
        state.degraded_rows += 1;
        state.degraded = true;
    }
}

fn record_degradation_if_pure(inner: &Inner, producer: AuditProducer) {
    if classify(producer) == AuditProductionClass::PureObservability {
        inner.record_degradation();
    }
}

/// Armed for the lifetime of one generation's supervision. If dropped while
/// still armed — the supervisor's own frame unwinding (panic) or being
/// dropped mid-await (cancellation/shutdown abort) — the guard fails the
/// generation before returning control to whatever tore it down, so the
/// caller never observes a background-task-count restoration that raced
/// ahead of the failure broadcast (R1).
struct SupervisorGuard<'a> {
    inner: &'a Arc<Inner>,
    waiting: Option<Vec<Waiting>>,
}

impl<'a> SupervisorGuard<'a> {
    fn armed(inner: &'a Arc<Inner>, waiting: Vec<Waiting>) -> Self {
        Self {
            inner,
            waiting: Some(waiting),
        }
    }

    /// Clean disarm: the caller has fully classified the outcome and taken
    /// ownership of the waiters to resolve them itself.
    fn disarm(mut self) -> Vec<Waiting> {
        self.waiting.take().unwrap_or_default()
    }

    /// Explicit failure classification reached without unwinding/cancelling
    /// this frame (e.g. a `JoinError` was returned normally). Equivalent to
    /// what `Drop` does when armed, but callable inline.
    fn fail(mut self, reason: AuditTerminalReason) {
        if let Some(waiting) = self.waiting.take() {
            self.inner.fail_driver(reason, waiting);
        }
    }
}

impl Drop for SupervisorGuard<'_> {
    fn drop(&mut self) {
        let Some(waiting) = self.waiting.take() else {
            return;
        };
        let reason = if std::thread::panicking() {
            AuditTerminalReason::DriverPanicked
        } else {
            AuditTerminalReason::DriverCancelled
        };
        self.inner.fail_driver(reason, waiting);
    }
}

enum RetryDecision {
    Retry,
    Terminal(AuditTerminalReason),
}

fn classify_store_error(err: &StorageError) -> RetryDecision {
    match err {
        StorageError::WriteQueueFull { .. } | StorageError::WriterTaskBusy { .. } => {
            RetryDecision::Retry
        }
        // Transient availability conditions, not judgments on the batch. The
        // events-daemon forwarding lane (ADR-170) reports an unreachable or
        // stalled daemon as `Pool`/`Timeout`; the direct SQL path reports
        // acquisition pressure the same way. Both are exactly what the
        // configured bounded retries exist for — treating them as terminal
        // would abandon a generation on the first blip of a daemon restart.
        StorageError::Pool { .. } | StorageError::Timeout { .. } => RetryDecision::Retry,
        StorageError::WriterTaskRequestFailed { request_state, .. }
        | StorageError::WriterTaskTerminated { request_state } => match request_state {
            WriterTaskRequestState::NotStarted | WriterTaskRequestState::TransactionRolledBack => {
                RetryDecision::Retry
            }
            WriterTaskRequestState::SideEffectsUnknown => RetryDecision::Retry,
        },
        StorageError::Unsupported { operation, .. }
            if operation.as_ref() == "append_events_idempotent" =>
        {
            RetryDecision::Terminal(AuditTerminalReason::IdempotencyUnsupported)
        }
        _ => RetryDecision::Terminal(AuditTerminalReason::StoreFailure),
    }
}

enum GenerationResult {
    Committed(Vec<EventAppendDisposition>),
    Failed(AuditTerminalReason),
    /// Fault-injection only: proves the supervisor's post-`Ok` consistency
    /// check actually runs.
    #[cfg_attr(not(any(test, feature = "fault-injection")), allow(dead_code))]
    FakedInconsistent,
}

/// The driver's own bound on how long `supervisor_loop` waits for one
/// generation's `run_generation` child task before abandoning it
/// (khive#2331; see [`AuditTerminalReason::DriverAppendAbandoned`]).
///
/// Derived from `resolution_deadline` rather than a second config field, at
/// 3x it. That margin is load-bearing, not arbitrary: `AuditBatch::new`
/// already enforces `resolution_deadline >= admission_deadline`, so the
/// worst-case *caller*-side wait for a `submit_until_resolved` row —
/// `admission_deadline` then `resolution_deadline` in sequence — is always
/// `<= 2 * resolution_deadline`. Bounding the driver at `3 *
/// resolution_deadline` guarantees it can never resolve a still-waiting
/// caller's row with `DriverAppendAbandoned` ahead of that caller's own,
/// more specific `ResolutionDeadlineExpired` reason.
///
/// This bound alone only stops the driver from holding one stalled
/// generation forever — a store whose append never returns still mints one
/// detached append per `driver_append_deadline` with nothing capping how
/// many run at once. `AuditBatchConfig::max_abandoned_appends` closes that:
/// once that many detached appends are outstanding, further generations are
/// shed with [`AuditTerminalReason::StoreWedged`] instead of attempting
/// another append. Combined, the two bounds guarantee at most
/// `max_abandoned_appends` live detached appends at any time, at most that
/// many retained event batches, and a shed generation costs no store work
/// at all — it resolves inside this function's caller without ever calling
/// `run_generation`.
fn driver_append_deadline(config: &AuditBatchConfig) -> Duration {
    config.resolution_deadline.saturating_mul(3)
}

async fn run_generation(
    store: Arc<dyn EventStore>,
    events: Vec<Event>,
    config: Arc<AuditBatchConfig>,
) -> GenerationResult {
    #[cfg(any(test, feature = "fault-injection"))]
    if fault::CHILD_PANIC.swap(false, Ordering::SeqCst) {
        panic!("adr133 fault injection: audit_batch child_panic");
    }
    #[cfg(any(test, feature = "fault-injection"))]
    if fault::INCONSISTENT_EXIT.swap(false, Ordering::SeqCst) {
        return GenerationResult::FakedInconsistent;
    }

    let mut attempt: u8 = 0;
    loop {
        attempt += 1;
        match store.append_events_idempotent(events.clone()).await {
            Ok(result) => return GenerationResult::Committed(result.rows),
            Err(err) => match classify_store_error(&err) {
                RetryDecision::Retry if attempt < config.max_commit_attempts.get() => {
                    tokio::time::sleep(config.retry_backoff).await;
                }
                RetryDecision::Retry => {
                    return GenerationResult::Failed(AuditTerminalReason::RetryExhausted)
                }
                RetryDecision::Terminal(reason) => return GenerationResult::Failed(reason),
            },
        }
    }
}

/// The batch owner. Constructed once per configured `EventStore`; every
/// dispatch-audit call site routes its row through [`AuditBatch::submit`]
/// instead of taking its own writer-task acquisition.
pub struct AuditBatch {
    inner: Arc<Inner>,
    store: Arc<dyn EventStore>,
    config: Arc<AuditBatchConfig>,
    supervisor: Mutex<Option<JoinHandle<()>>>,
}

impl AuditBatch {
    pub fn new(store: Arc<dyn EventStore>, config: AuditBatchConfig) -> Arc<Self> {
        debug_assert!(
            config.resolution_deadline >= config.admission_deadline,
            "resolution_deadline ({:?}) must be at least admission_deadline ({:?})",
            config.resolution_deadline,
            config.admission_deadline
        );
        Arc::new(Self {
            inner: Arc::new(Inner {
                state: Mutex::new(State::new()),
            }),
            store,
            config: Arc::new(config),
            supervisor: Mutex::new(None),
        })
    }

    /// Process-lifetime audit-batch health counters, for the registry/
    /// runtime owner to feed into `db_diagnostics` (D8's operator surface).
    /// Unlike `test_internals::AuditBatchSnapshot::metrics_snapshot`,
    /// which is test-only (the module is cfg-gated and invisible to the
    /// default doc build, so an intra-doc link cannot resolve), this is
    /// always available.
    pub fn health_metrics(&self) -> AuditBatchHealthMetrics {
        let state = self.inner.state.lock();
        AuditBatchHealthMetrics {
            flush_failures: state.flush_failures,
            degraded_rows: state.degraded_rows,
            degraded: state.degraded,
            late_append_commits: state.late_append_commits,
            late_append_failures: state.late_append_failures,
        }
    }

    fn spawn_supervisor_if_idle(&self) {
        let inner = self.inner.clone();
        let store = self.store.clone();
        let config = self.config.clone();
        let handle = tokio::spawn(async move {
            supervisor_loop(inner, store, config).await;
        });
        *self.supervisor.lock() = Some(handle);
    }

    /// Enqueue one row and keep waiting for its real generation outcome past
    /// the ordinary admission wait deadline, up to `resolution_deadline`.
    ///
    /// This is the narrow khive#2256 seam for a successful operation whose
    /// domain effect has already committed. Returning
    /// [`AuditTerminalReason::AdmissionDeadlineExpired`] there would report a
    /// false operation failure and invite an unsafe retry while the same
    /// audit row remains enqueued. Pre-enqueue refusal and genuine terminal
    /// generation failures still return normally. The post-admission wait is
    /// itself bounded by `resolution_deadline`
    /// ([`AuditTerminalReason::ResolutionDeadlineExpired`]) so a stalled
    /// store cannot retain this caller, its request slot, and its audit-lane
    /// waiter forever (khive#2331).
    ///
    /// The two deadlines therefore mean different things to the caller, and
    /// callers key on the difference. Admission expiry means the row is
    /// enqueued and this seam keeps waiting; the generation commits it
    /// independently, so a dispatch that reaches it reports its committed
    /// result and counts the row as unresolved. Resolution expiry means the
    /// caller waited for the real outcome and never received one: the commit
    /// is unconfirmed, not proven absent. That is returned as a terminal
    /// reason and the dispatch propagates it as a structured
    /// committed-outcome error carrying the domain result, with no retryable
    /// context, for every verb rather than only for the receipt verb. Never
    /// replay the handler or re-enqueue the row on it; the original driver
    /// work continues on its own.
    pub(crate) async fn submit_until_resolved(
        &self,
        row: PreparedAuditRow,
    ) -> Result<AuditCommitOutcome, AuditTerminalReason> {
        self.submit_with_wait_policy(row, true).await
    }

    async fn submit_with_wait_policy(
        &self,
        row: PreparedAuditRow,
        wait_until_resolved: bool,
    ) -> Result<AuditCommitOutcome, AuditTerminalReason> {
        // Pre-enqueue validation (invariant 3): a malformed row is rejected
        // before it can share a generation with anyone else's.
        if self.store.preflight_event(&row.event).is_err() {
            return Err(AuditTerminalReason::PreflightRejected);
        }

        let producer = row.producer;
        let (tx, mut rx) = oneshot::channel();
        let need_spawn = {
            let mut state = self.inner.state.lock();
            match state.lifecycle {
                Lifecycle::Closed | Lifecycle::Closing => {
                    return Err(AuditTerminalReason::AdmissionClosed)
                }
                Lifecycle::Failed(reason) => return Err(reason),
                Lifecycle::Open => {}
            }
            if state.pending.len() >= self.config.max_pending_rows.get() {
                return Err(AuditTerminalReason::QueueAdmissionExhausted);
            }
            state.pending.push(Waiting {
                event: row.event,
                producer,
                responder: tx,
            });
            state.submitted_rows += 1;
            let need_spawn = !state.driver_active;
            if need_spawn {
                state.driver_active = true;
            }
            need_spawn
        };
        if need_spawn {
            self.spawn_supervisor_if_idle();
        }

        if wait_until_resolved {
            match tokio::time::timeout(self.config.admission_deadline, &mut rx).await {
                Ok(Ok(result)) => return result,
                Ok(Err(_recv_error)) => return Err(AuditTerminalReason::DriverJoinLost),
                Err(_elapsed) => tracing::warn!(
                    ?producer,
                    "strict audit obligation remains enqueued after the admission wait deadline; \
                     waiting up to the resolution deadline for its real terminal outcome"
                ),
            }
            return match tokio::time::timeout(self.config.resolution_deadline, rx).await {
                Ok(Ok(result)) => result,
                Ok(Err(_recv_error)) => Err(AuditTerminalReason::DriverJoinLost),
                // Same non-removal contract as the `AdmissionDeadlineExpired`
                // arm below: the row is left exactly where the driver holds
                // it. The caller's domain effect already committed by the
                // time it reached this wait, so it is never retried here;
                // the audit outcome itself is what remains unresolved.
                Err(_elapsed) => {
                    tracing::warn!(
                        ?producer,
                        "strict audit obligation remains unresolved after the resolution \
                         deadline; giving up on this caller's wait — the committed effect is \
                         not retried and the row is not re-enqueued"
                    );
                    Err(AuditTerminalReason::ResolutionDeadlineExpired)
                }
            };
        }

        match tokio::time::timeout(self.config.admission_deadline, rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_recv_error)) => Err(AuditTerminalReason::DriverJoinLost),
            // The row was already pushed onto `state.pending` above (and
            // `submitted_rows` incremented) before this wait began — this is
            // a deadline elapsing on an enqueued row, not a queue-full
            // refusal, so it gets its own terminal reason (khive#2117,
            // khive#2208). The row is left in place for the driver to drain;
            // this arm performs no removal.
            Err(_elapsed) => Err(AuditTerminalReason::AdmissionDeadlineExpired),
        }
    }
}

async fn supervisor_loop(
    inner: Arc<Inner>,
    store: Arc<dyn EventStore>,
    config: Arc<AuditBatchConfig>,
) {
    loop {
        let (waiting, wedged_generation_id) = {
            let mut state = inner.state.lock();
            if matches!(state.lifecycle, Lifecycle::Failed(_)) {
                state.driver_active = false;
                break;
            }
            if state.pending.is_empty() {
                state.driver_active = false;
                break;
            }
            let take_n = state
                .pending
                .len()
                .min(config.max_rows_per_generation.get());
            let waiting: Vec<Waiting> = state.pending.drain(..take_n).collect();
            let generation_id = state.next_generation_id;
            state.next_generation_id += 1;
            // khive#2331: a store whose append never returns must not mint
            // an unbounded number of detached appends. Before committing to
            // spawning this generation's child, check whether the cap is
            // already saturated by prior abandoned generations still
            // running in the background — if so, this generation is shed
            // below instead of ever calling the store.
            if state.outstanding_abandoned_appends >= config.max_abandoned_appends.get() {
                (waiting, Some(generation_id))
            } else {
                state.in_flight_generation = Some(generation_id);
                (waiting, None)
            }
        };

        if let Some(generation_id) = wedged_generation_id {
            tracing::warn!(
                generation_id,
                max_abandoned_appends = config.max_abandoned_appends.get(),
                "audit store treated as wedged: max_abandoned_appends detached appends are \
                 already outstanding; shedding this generation without attempting an append"
            );
            let submitted = waiting.len() as u64;
            {
                let mut state = inner.state.lock();
                state.flush_failures += 1;
                state.generations.push(AuditGenerationSnapshot {
                    generation_id,
                    submitted_rows: submitted,
                    committed_rows: 0,
                    store_batch_calls: 0,
                    terminal_reason: Some(AuditTerminalReason::StoreWedged),
                });
            }
            for w in waiting {
                record_degradation_if_pure(&inner, w.producer);
                let _ = w.responder.send(Err(AuditTerminalReason::StoreWedged));
            }
            continue;
        }

        #[cfg(any(test, feature = "fault-injection"))]
        if fault::SUPERVISOR_PANIC.swap(false, Ordering::SeqCst) {
            let _guard = SupervisorGuard::armed(&inner, waiting);
            panic!("adr133 fault injection: audit_batch supervisor_panic");
        }
        #[cfg(any(test, feature = "fault-injection"))]
        if fault::SUPERVISOR_SLEEP_BEFORE_SPAWN.swap(false, Ordering::SeqCst) {
            let guard = SupervisorGuard::armed(&inner, waiting);
            tokio::time::sleep(Duration::from_secs(3600)).await;
            drop(guard);
            continue;
        }

        let guard = SupervisorGuard::armed(&inner, waiting);
        let events: Vec<Event> = guard
            .waiting
            .as_ref()
            .expect("guard freshly armed")
            .iter()
            .map(|w| w.event.clone())
            .collect();

        let mut child: JoinHandle<GenerationResult> =
            tokio::spawn(run_generation(store.clone(), events, config.clone()));

        #[cfg(any(test, feature = "fault-injection"))]
        if fault::CHILD_CANCEL.swap(false, Ordering::SeqCst) {
            child.abort();
        }
        #[cfg(any(test, feature = "fault-injection"))]
        if fault::JOIN_LOST.swap(false, Ordering::SeqCst) {
            drop(child);
            guard.fail(AuditTerminalReason::DriverJoinLost);
            continue;
        }

        let append_deadline = driver_append_deadline(&config);
        let join_result = match tokio::time::timeout(append_deadline, &mut child).await {
            Ok(join_result) => join_result,
            Err(_elapsed) => {
                // The append is still in flight and may not be safely
                // cancellable mid-write (a blocking storage call cannot be
                // forced to stop), so `child` is not aborted here — it is
                // handed to a detached reaper that drives it to completion
                // and discards whatever it eventually returns. Every waiter
                // on this generation is resolved now, and the driver loops
                // back to `pending` immediately instead of staying pinned on
                // this one stalled call, which is what actually starves
                // later admission (khive#2331).
                tracing::warn!(
                    ?append_deadline,
                    "audit generation append exceeded the driver's own bound; abandoning \
                     this generation so admission capacity recovers for later callers"
                );
                let waiting = guard.disarm();
                let submitted = waiting.len() as u64;
                {
                    let mut state = inner.state.lock();
                    state.in_flight_generation = None;
                    state.flush_failures += 1;
                    state.outstanding_abandoned_appends += 1;
                    let generation_id = state.next_generation_id.saturating_sub(1);
                    state.generations.push(AuditGenerationSnapshot {
                        generation_id,
                        submitted_rows: submitted,
                        committed_rows: 0,
                        store_batch_calls: 0,
                        terminal_reason: Some(AuditTerminalReason::DriverAppendAbandoned),
                    });
                }
                for w in waiting {
                    record_degradation_if_pure(&inner, w.producer);
                    let _ = w
                        .responder
                        .send(Err(AuditTerminalReason::DriverAppendAbandoned));
                }
                let reaper_inner = inner.clone();
                tokio::spawn(async move {
                    let result = child.await;
                    let mut state = reaper_inner.state.lock();
                    state.outstanding_abandoned_appends =
                        state.outstanding_abandoned_appends.saturating_sub(1);
                    match result {
                        Ok(GenerationResult::Committed(_)) => state.late_append_commits += 1,
                        Ok(GenerationResult::Failed(_) | GenerationResult::FakedInconsistent)
                        | Err(_) => state.late_append_failures += 1,
                    }
                });
                continue;
            }
        };
        match join_result {
            Err(join_err) => {
                let reason = if join_err.is_panic() {
                    AuditTerminalReason::DriverPanicked
                } else {
                    AuditTerminalReason::DriverCancelled
                };
                guard.fail(reason);
            }
            Ok(GenerationResult::FakedInconsistent) => {
                guard.fail(AuditTerminalReason::DriverExitedInconsistent);
            }
            Ok(GenerationResult::Failed(reason)) => {
                let waiting = guard.disarm();
                let submitted = waiting.len() as u64;
                {
                    let mut state = inner.state.lock();
                    state.in_flight_generation = None;
                    state.store_batch_calls += 1;
                    state.flush_failures += 1;
                    state.degraded = true;
                    let generation_id = state.next_generation_id.saturating_sub(1);
                    state.generations.push(AuditGenerationSnapshot {
                        generation_id,
                        submitted_rows: submitted,
                        committed_rows: 0,
                        store_batch_calls: 1,
                        terminal_reason: Some(reason),
                    });
                }
                for w in waiting {
                    record_degradation_if_pure(&inner, w.producer);
                    let _ = w.responder.send(Err(reason));
                }
            }
            Ok(GenerationResult::Committed(dispositions)) => {
                let waiting = guard.disarm();
                if dispositions.len() != waiting.len() {
                    // Defensive: the store must preserve input order/length.
                    // Treat a mismatch as the driver having exited in a
                    // state that cannot be reconciled with the accepted
                    // waiters.
                    {
                        let mut state = inner.state.lock();
                        state.in_flight_generation = None;
                    }
                    inner.fail_driver(AuditTerminalReason::DriverExitedInconsistent, waiting);
                    continue;
                }
                let mut committed_n = 0u64;
                for d in &dispositions {
                    if !matches!(d, EventAppendDisposition::IdentityConflict) {
                        committed_n += 1;
                    }
                }
                let submitted = dispositions.len() as u64;
                {
                    let mut state = inner.state.lock();
                    state.in_flight_generation = None;
                    state.store_batch_calls += 1;
                    state.committed_rows += committed_n;
                    let generation_id = state.next_generation_id.saturating_sub(1);
                    state.generations.push(AuditGenerationSnapshot {
                        generation_id,
                        submitted_rows: submitted,
                        committed_rows: committed_n,
                        store_batch_calls: 1,
                        terminal_reason: None,
                    });
                }
                for (w, disposition) in waiting.into_iter().zip(dispositions) {
                    let result = match disposition {
                        EventAppendDisposition::Inserted => Ok(AuditCommitOutcome::Committed),
                        EventAppendDisposition::AlreadyPresentIdentical => {
                            Ok(AuditCommitOutcome::AlreadyPresentIdentical)
                        }
                        EventAppendDisposition::IdentityConflict => {
                            record_degradation_if_pure(&inner, w.producer);
                            Err(AuditTerminalReason::IdentityConflict)
                        }
                    };
                    let _ = w.responder.send(result);
                }
            }
        }
    }
}

#[async_trait::async_trait]
impl AuditBatchControl for AuditBatch {
    async fn submit(
        &self,
        row: PreparedAuditRow,
    ) -> Result<AuditCommitOutcome, AuditTerminalReason> {
        self.submit_with_wait_policy(row, false).await
    }

    async fn quiesce(&self) -> Result<(), AuditTerminalReason> {
        loop {
            let (lifecycle, idle) = {
                let state = self.inner.state.lock();
                (state.lifecycle.clone(), state.is_idle())
            };
            match lifecycle {
                Lifecycle::Failed(reason) => return Err(reason),
                Lifecycle::Closed => return Ok(()),
                _ if idle => return Ok(()),
                _ => tokio::time::sleep(Duration::from_millis(2)).await,
            }
        }
    }

    async fn close_and_drain(&self) -> Result<(), AuditTerminalReason> {
        {
            let mut state = self.inner.state.lock();
            if matches!(state.lifecycle, Lifecycle::Open) {
                state.lifecycle = Lifecycle::Closing;
            }
        }
        let result = AuditBatchControl::quiesce(self).await;
        {
            let mut state = self.inner.state.lock();
            if result.is_ok() && matches!(state.lifecycle, Lifecycle::Closing) {
                state.lifecycle = Lifecycle::Closed;
            }
        }
        let handle = self.supervisor.lock().take();
        if let Some(handle) = handle {
            let _ = handle.await;
        }
        result
    }
}

#[cfg(any(test, feature = "fault-injection"))]
mod fault {
    use std::sync::atomic::AtomicBool;

    pub(super) static CHILD_PANIC: AtomicBool = AtomicBool::new(false);
    pub(super) static CHILD_CANCEL: AtomicBool = AtomicBool::new(false);
    pub(super) static SUPERVISOR_PANIC: AtomicBool = AtomicBool::new(false);
    pub(super) static SUPERVISOR_SLEEP_BEFORE_SPAWN: AtomicBool = AtomicBool::new(false);
    pub(super) static JOIN_LOST: AtomicBool = AtomicBool::new(false);
    pub(super) static INCONSISTENT_EXIT: AtomicBool = AtomicBool::new(false);
}

/// Deterministic fault-injection arms consumed by exactly one subsequent
/// generation each. Test-only surface, gated identically to the rest of this
/// crate's `fault-injection` fixtures (see `crate::operations`).
#[cfg(any(test, feature = "fault-injection"))]
pub mod fault_injection {
    use std::sync::atomic::Ordering;

    pub fn arm_child_panic() {
        super::fault::CHILD_PANIC.store(true, Ordering::SeqCst);
    }
    pub fn arm_child_cancel() {
        super::fault::CHILD_CANCEL.store(true, Ordering::SeqCst);
    }
    pub fn arm_supervisor_panic() {
        super::fault::SUPERVISOR_PANIC.store(true, Ordering::SeqCst);
    }
    pub fn arm_supervisor_sleep_before_spawn() {
        super::fault::SUPERVISOR_SLEEP_BEFORE_SPAWN.store(true, Ordering::SeqCst);
    }
    pub fn arm_join_lost() {
        super::fault::JOIN_LOST.store(true, Ordering::SeqCst);
    }
    pub fn arm_inconsistent_exit() {
        super::fault::INCONSISTENT_EXIT.store(true, Ordering::SeqCst);
    }
}

// ── Test-attribution surface (R2) ────────────────────────────────────────
//
// Additive only: production attribution still lands in the public
// `db_diagnostics` counters via `metrics_snapshot`-shaped data supplied at
// that seam. This surface never substitutes for it.
#[cfg(any(test, feature = "test-internals"))]
mod test_internals {
    use super::*;

    /// A point-in-time view of the batch's counters and generation history,
    /// used by mechanism tests to compute a delta across a measured
    /// operation.
    #[derive(Debug, Clone)]
    pub struct AuditBatchSnapshot {
        pub pending_rows: usize,
        pub in_flight_generation: Option<u64>,
        pub driver_active: bool,
        pub next_generation_id: u64,
        pub submitted_rows: u64,
        pub committed_rows: u64,
        pub store_batch_calls: u64,
        pub per_generation: Vec<AuditGenerationSnapshot>,
        /// Live count of detached `DriverAppendAbandoned` appends still
        /// running, bounded by `AuditBatchConfig::max_abandoned_appends`.
        pub outstanding_abandoned_appends: usize,
    }

    impl AuditBatchSnapshot {
        pub fn is_idle(&self) -> bool {
            self.pending_rows == 0 && self.in_flight_generation.is_none() && !self.driver_active
        }
    }

    #[derive(Debug, Clone, Copy)]
    pub struct AuditBatchMetricsSnapshot {
        pub flush_failures: u64,
        pub degraded_rows: u64,
        pub degraded: bool,
        pub late_append_commits: u64,
        pub late_append_failures: u64,
    }

    #[derive(Debug, Clone, PartialEq)]
    pub struct AuditBatchDelta {
        pub submitted_rows: u64,
        pub committed_rows: u64,
        pub store_batch_calls: u64,
        pub per_generation: Vec<AuditGenerationSnapshot>,
    }

    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    pub enum AuditSnapshotError {
        CounterRegressed,
        GenerationHistoryRegressed,
    }

    impl AuditBatch {
        pub fn test_snapshot(&self) -> AuditBatchSnapshot {
            let state = self.inner.state.lock();
            AuditBatchSnapshot {
                pending_rows: state.pending.len(),
                in_flight_generation: state.in_flight_generation,
                driver_active: state.driver_active,
                next_generation_id: state.next_generation_id,
                submitted_rows: state.submitted_rows,
                committed_rows: state.committed_rows,
                store_batch_calls: state.store_batch_calls,
                per_generation: state.generations.clone(),
                outstanding_abandoned_appends: state.outstanding_abandoned_appends,
            }
        }

        pub fn metrics_snapshot(&self) -> AuditBatchMetricsSnapshot {
            let state = self.inner.state.lock();
            AuditBatchMetricsSnapshot {
                flush_failures: state.flush_failures,
                degraded_rows: state.degraded_rows,
                degraded: state.degraded,
                late_append_commits: state.late_append_commits,
                late_append_failures: state.late_append_failures,
            }
        }

        /// Abort the currently-retained supervisor `JoinHandle`, if any,
        /// simulating a shutdown abort landing mid-generation. Returns
        /// whether a handle was found and aborted. Test-only: exercises the
        /// R1 supervisor-cancellation path without reaching into a private
        /// field from outside this module.
        pub fn test_abort_supervisor(&self) -> bool {
            let handle = self.supervisor.lock().take();
            match handle {
                Some(handle) => {
                    handle.abort();
                    true
                }
                None => false,
            }
        }
    }

    /// Checked monotonic subtraction. Rejects a regressed counter or a
    /// generation history that is not an append-only extension of `before`.
    pub fn audit_delta(
        before: &AuditBatchSnapshot,
        after: &AuditBatchSnapshot,
    ) -> Result<AuditBatchDelta, AuditSnapshotError> {
        let submitted_rows = after
            .submitted_rows
            .checked_sub(before.submitted_rows)
            .ok_or(AuditSnapshotError::CounterRegressed)?;
        let committed_rows = after
            .committed_rows
            .checked_sub(before.committed_rows)
            .ok_or(AuditSnapshotError::CounterRegressed)?;
        let store_batch_calls = after
            .store_batch_calls
            .checked_sub(before.store_batch_calls)
            .ok_or(AuditSnapshotError::CounterRegressed)?;
        if after.per_generation.len() < before.per_generation.len() {
            return Err(AuditSnapshotError::GenerationHistoryRegressed);
        }
        if after.per_generation[..before.per_generation.len()] != before.per_generation[..] {
            return Err(AuditSnapshotError::GenerationHistoryRegressed);
        }
        let per_generation = after.per_generation[before.per_generation.len()..].to_vec();
        Ok(AuditBatchDelta {
            submitted_rows,
            committed_rows,
            store_batch_calls,
            per_generation,
        })
    }

    /// Exhaustive, non-wildcard producer classification (see
    /// [`super::AuditProducer`] and [`super::classify`], the real
    /// crate-private definitions this mirrors one-for-one). The doctest
    /// below proves the general property those definitions rely on: a match
    /// over a non-`#[non_exhaustive]` enum that omits a variant, with no
    /// wildcard arm to silently absorb it, fails to compile rather than
    /// passing an incomplete classification.
    ///
    /// ```compile_fail
    /// enum AuditProducer {
    ///     GateDenied,
    ///     DispatchSucceeded,
    ///     DispatchFailed,
    ///     UnknownVerb,
    ///     GitDigestReceipt,
    ///     ConfigLocked,
    ///     RecallExecuted,
    /// }
    ///
    /// fn describe(p: AuditProducer) -> &'static str {
    ///     match p {
    ///         AuditProducer::GateDenied => "obligation",
    ///         AuditProducer::DispatchSucceeded => "obligation",
    ///         AuditProducer::DispatchFailed => "obligation",
    ///         AuditProducer::UnknownVerb => "obligation",
    ///         AuditProducer::GitDigestReceipt => "obligation",
    ///         AuditProducer::ConfigLocked => "observability",
    ///         // RecallExecuted intentionally omitted: a non-exhaustive
    ///         // match must fail to compile, proving no variant can
    ///         // silently fall through an absent wildcard arm.
    ///     }
    /// }
    /// ```
    #[allow(dead_code)]
    struct DoctestAnchor;
}

#[cfg(any(test, feature = "test-internals"))]
pub use test_internals::*;