car_server_core/

feedback_drain.rs

//! The feedback-spool drain — the background loop that uploads durable
//! feedback submissions when connectivity and server capability allow.
//!
//! PR A, U5 of `docs/plans/2026-08-31-car-feedback-system.md` (Architecture
//! DAEMON "Spool" drain; binding requirements 3/12/13/30/31); outcomes
//! SPOOL-1/2/5/6/7, DEG-1/3, PRIV-1, IDEM-1.
//!
//! # Event-driven with a local-only park tick (PRIV-1 / SPOOL-5)
//!
//! Unlike the daemon's other background loops (`registry_reaper`,
//! `command_scheduler` — interval tickers), this loop makes NO network on a
//! schedule. It parks on a [`tokio::sync::Notify`] until `feedback.submit`
//! wakes it ([`wake_feedback_drain`]), runs drain rounds while the outbox has
//! work, and goes back to sleep. With an empty outbox there is no network
//! call, nothing: [`run_drain_round`] returns before constructing a single
//! request — before even consulting the transport — when no Queued entry
//! exists. The distinguishing test
//! `empty_outbox_produces_no_transport_calls` pins that with a counting mock,
//! and `spawned_drain_parks_idle_and_drains_on_wake` pins the park phase.
//!
//! The in-process `Notify` cannot see a CLI's direct spool enqueue (the CLI
//! writes to the spool with the daemon idle — codex finding #11), so the park
//! additionally wakes every [`DrainConfig::park_check_interval`] for a
//! **LOCAL-ONLY** disk check: does the outbox hold a PENDING (Queued/Sending)
//! entry? When empty — the overwhelmingly common case — the tick reads the
//! directory listing only; when entries exist it asks the spool which are
//! still pending, so settled entries waiting out their TTL never break the
//! park. Either way the tick touches the filesystem only and NEVER the
//! transport, so PRIV-1 holds; when a pending entry appeared, the loop runs a
//! normal drain round (which itself consults the transport only for Queued
//! entries). The distinguishing test
//! `cli_direct_enqueue_while_parked_drains_within_one_tick` enqueues through
//! a second `Spool` handle without calling the drain handle.
//!
//! The one boot-time action is a single local spool read (plus one round IF
//! entries were spooled while the daemon was down — SPOOL-2's restart half);
//! on an idle install that read finds nothing and the loop parks.
//!
//! # Per-entry protocol (requirement 31 / SPOOL-6, checklist #1)
//!
//! `mark_sending` → transport → apply the returned
//! [`DrainAction`] to the spool: `Acknowledge → mark_acknowledged`,
//! `Requeue/RequeueAfter → mark_queued`, terminal → `mark_terminal`. Every
//! transition is a PERSISTED `state.json` rewrite in
//! `car_feedback_core::spool`, and the tests below re-open the spool from
//! disk to assert the durable state, not an in-memory mirror. An entry found
//! `Sending` at round start is a crashed drain's leftover: it is recovered to
//! Queued and re-sent under the same immutable `client_submission_id` — safe
//! by IDEM-1 (the server's idempotent 200 replay) and never a duplicate mint.
//!
//! # Pacing and backoff (requirement 12 / SPOOL-7)
//!
//! At most one upload per [`DrainConfig::min_upload_interval`] (15s),
//! `Retry-After` honored up to [`DrainConfig::max_backoff`] (the transport
//! already clamps the header to its `MAX_RETRY_AFTER_SECS`; the loop clamps
//! again so no single header can park the drain until restart) and waited out
//! through the same interruptible `wait_or_wake` as every other wait — a
//! fresh submit's wake runs a round, whose upload the 15s pacing still
//! spaces, and a repeat 429 re-arms the wait. This pacing is strictly PER-DEVICE: N
//! reconnecting devices still submit N/15s in aggregate, so it cannot keep a
//! CAR fleet inside the server's per-org `car-feedback` sliding window on its
//! own — the fleet-wide no-429 guarantee is delivered server-side by that
//! dedicated rate-limit policy (CAR's OWN bucket on `CarFeedbackController`,
//! never the shared bug-report policy).
//! Retriable failures back off exponentially up to
//! [`DrainConfig::max_backoff`]. Held entries (no session / the capability
//! probe not advertising the lane) re-check on
//! [`DrainConfig::held_recheck_interval`] — that is what lets DEG-3's
//! capability-flip drain happen without a user action, and it runs only
//! while the outbox is non-empty, so PRIV-1 holds.
//!
//! # Retriable-attempt cap (finding F13) — per process, never terminal
//!
//! 5xx / 408 / 3xx / network failures are retriable by taxonomy, and without
//! a ceiling one entry the server keeps failing would be re-posted every
//! `max_backoff` for the life of the daemon. The spool's `state.json` carries
//! no attempt counter (its record is immutable after enqueue except
//! `state`/`settled_at`), and no HONEST terminal state exists for "gave up on
//! a retriable failure": `TerminalRejected` means the server rejected the
//! report and `TerminalActionable` names a user action. So the cap is an
//! in-memory [`RetryLedger`] owned by the loop: after
//! [`DrainConfig::max_retriable_attempts`] consecutive retriable failures an
//! entry is skipped by this process's further rounds — BEFORE the eligibility
//! probe, so it costs no network — logged once, and left `Queued` on disk
//! (visible through SPOOL-3's staleness notice, exportable, never silently
//! converted into a report nobody sees — requirement 13). The budget renews
//! on daemon restart. A durable dead-letter needs a spool schema change
//! (attempt counter + an honest terminal variant) and is a recorded follow-up.
//! 429 (`Retry-After`) and holds are not failures of the entry and never
//! consume budget.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use std::time::Duration;

use async_trait::async_trait;
use tokio::sync::Notify;
use tokio::time::Instant;

use car_feedback_core::spool::{
    IdentityLane, Spool, SpoolEntryId, SpoolEntrySummary, SpoolState, TerminalReason,
};
use car_parslee::feedback_transport::{
    DrainAction, DrainEligibility, FeedbackTransport, FeedbackTransportError, SubmitOutcome,
    TransportActionableReason,
};

use crate::feedback::FEEDBACK_OUTBOX_DIR;
use crate::session::ServerState;

/// Tuning for one drain loop. Production uses [`DrainConfig::default`]; tests
/// construct short intervals (a code-level parameter, not a knob — there is
/// no env var and the default is always right, per house rule #1a).
#[derive(Debug, Clone)]
pub struct DrainConfig {
    /// SPOOL-7: minimum gap between uploads (shared org rate-limit budget).
    pub min_upload_interval: Duration,
    /// First retry delay for a retriable failure; doubles per failed round.
    pub initial_backoff: Duration,
    /// Ceiling for the exponential schedule.
    pub max_backoff: Duration,
    /// How often held entries (capability not advertised / no session /
    /// anonymous) re-check —
    /// only while the outbox is non-empty (DEG-3 without violating PRIV-1).
    pub held_recheck_interval: Duration,
    /// While PARKED (no pending entries), how often the loop does the
    /// LOCAL-ONLY disk check for pending entries that arrived without a
    /// `Notify` — a CLI's direct spool enqueue (finding #11). The check reads
    /// the outbox (the directory listing, then the spool's own list when
    /// entries exist) and nothing else: no transport call is ever made from
    /// the park tick itself.
    pub park_check_interval: Duration,
    /// Finding F13: consecutive retriable failures (5xx/408/3xx/network) an
    /// entry may accumulate in THIS process before the drain stops re-posting
    /// it — the entry stays `Queued` on disk and gets a fresh budget on the
    /// next daemon start (see the module docs). At the default schedule the
    /// default of 24 is roughly five hours of retrying (30s doubling to the
    /// 15-minute ceiling, then 18 more 15-minute rounds).
    pub max_retriable_attempts: u32,
}

impl Default for DrainConfig {
    fn default() -> Self {
        DrainConfig {
            min_upload_interval: Duration::from_secs(15),
            initial_backoff: Duration::from_secs(30),
            max_backoff: Duration::from_secs(15 * 60),
            held_recheck_interval: Duration::from_secs(15 * 60),
            park_check_interval: Duration::from_secs(60),
            max_retriable_attempts: 24,
        }
    }
}

/// The per-process retry ledger (finding F13): consecutive retriable failures
/// per queued entry, owned by one drain loop and threaded through every
/// [`run_drain_round`]. Deliberately NOT durable and NOT a terminal
/// transition — see the module docs for why. Rows for entries that settled
/// or were pruned are dropped each round, so the map never outgrows the live
/// outbox.
#[derive(Debug, Default)]
pub struct RetryLedger {
    failures: HashMap<SpoolEntryId, u32>,
}

impl RetryLedger {
    /// One more retriable failure for `id`; logs ONCE, at the moment the cap
    /// is reached, with the user-actionable recovery.
    fn record_failure(&mut self, id: &SpoolEntryId, cap: u32) {
        let count = self.failures.entry(id.clone()).or_insert(0);
        *count = count.saturating_add(1);
        if *count == cap {
            tracing::warn!(
                target: "car::feedback",
                entry = %id, attempts = cap,
                "feedback entry hit this daemon's retriable-attempt cap; it stays queued \
                 (export it with `car feedback --export`) and retries again after the \
                 daemon restarts"
            );
        }
    }

    /// Has `id` used up this process's budget?
    fn is_exhausted(&self, id: &SpoolEntryId, cap: u32) -> bool {
        self.failures.get(id).is_some_and(|count| *count >= cap)
    }

    /// Keep only the rows for entries still queued — settled/pruned entries
    /// leave no residue.
    fn retain_queued(&mut self, queued: &[SpoolEntrySummary]) {
        self.failures
            .retain(|id, _| queued.iter().any(|entry| &entry.id == id));
    }
}

/// The drain's view of the upstream transport — exactly the two calls U5's
/// transport exposes, as a trait so tests count and script them.
#[async_trait]
pub trait DrainTransport: Send + Sync {
    async fn drain_eligible(&self, lane: &IdentityLane) -> DrainEligibility;
    async fn submit(
        &self,
        bundle: &car_feedback_core::bundle::RedactedBundle,
        lane: &IdentityLane,
        client_submission_id: &str,
    ) -> Result<SubmitOutcome, FeedbackTransportError>;
}

#[async_trait]
impl DrainTransport for FeedbackTransport {
    async fn drain_eligible(&self, lane: &IdentityLane) -> DrainEligibility {
        FeedbackTransport::drain_eligible(self, lane).await
    }
    async fn submit(
        &self,
        bundle: &car_feedback_core::bundle::RedactedBundle,
        lane: &IdentityLane,
        client_submission_id: &str,
    ) -> Result<SubmitOutcome, FeedbackTransportError> {
        FeedbackTransport::submit_report(self, bundle, lane, client_submission_id).await
    }
}

/// Boot-passive live transport (demand-driven credentials, car#661):
/// constructing [`FeedbackTransport`] resolves the API base, which falls
/// through to the published auth state — a SECRET-STORE read a cold daemon
/// must never perform (`cold_daemon_handshake_is_passive_but_explicit_status_reads`
/// pins the zero-read boot). The empty-spool round never consults the
/// transport (PRIV-1), so deferring construction to the first call keeps
/// boot, handshake, and idle parking credential-free; a construction failure
/// surfaces as a per-entry Hold with the reason instead of a never-started
/// drain.
struct LazyLiveTransport {
    inner: tokio::sync::OnceCell<FeedbackTransport>,
    #[cfg(test)]
    constructor_error: Option<String>,
}

impl LazyLiveTransport {
    fn new() -> Self {
        Self {
            inner: tokio::sync::OnceCell::new(),
            #[cfg(test)]
            constructor_error: None,
        }
    }

    #[cfg(test)]
    fn failing(error: &str) -> Self {
        Self {
            inner: tokio::sync::OnceCell::new(),
            constructor_error: Some(error.to_string()),
        }
    }

    async fn get(&self) -> Result<&FeedbackTransport, String> {
        #[cfg(test)]
        if let Some(error) = &self.constructor_error {
            return Err(error.clone());
        }
        self.inner
            .get_or_try_init(|| async { FeedbackTransport::live() })
            .await
    }
}

#[async_trait]
impl DrainTransport for LazyLiveTransport {
    async fn drain_eligible(&self, lane: &IdentityLane) -> DrainEligibility {
        match self.get().await {
            Ok(transport) => DrainTransport::drain_eligible(transport, lane).await,
            Err(error) => DrainEligibility::Hold {
                reason: format!("feedback transport unavailable: {error}"),
            },
        }
    }
    async fn submit(
        &self,
        bundle: &car_feedback_core::bundle::RedactedBundle,
        lane: &IdentityLane,
        client_submission_id: &str,
    ) -> Result<SubmitOutcome, FeedbackTransportError> {
        match self.get().await {
            Ok(transport) => {
                DrainTransport::submit(transport, bundle, lane, client_submission_id).await
            }
            // Unreachable in practice: every round consults drain_eligible
            // first, which Holds on a construction failure — but a sane
            // retriable error keeps the seam total.
            Err(error) => Err(FeedbackTransportError::FetchFailed(format!(
                "feedback transport unavailable: {error}"
            ))),
        }
    }
}

/// What one pass over the outbox concluded — drives the loop's next wait.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RoundOutcome {
    /// No Queued entries — park on the Notify (no timer, PRIV-1).
    Idle,
    /// Queued entries exist but every one is held (no session / capability
    /// not advertised /
    /// anonymous while `anonymousIntake` is false — or past this process's
    /// retriable-attempt cap, finding F13) — re-check on the held
    /// cadence (DEG-3).
    AllHeld,
    /// At least one entry settled (acknowledged or terminal) — run again
    /// immediately in case more remain.
    Progressed,
    /// Retriable failure(s) — wait out the exponential backoff, then retry.
    Backoff { min_delay: Duration },
    /// The server said 429 — no upload before this delay (Retry-After).
    RetryAfter(Duration),
}

/// Handle to a spawned drain: `wake()` after every `feedback.submit` enqueue.
#[derive(Clone)]
pub struct FeedbackDrainHandle {
    notify: Arc<Notify>,
}

impl FeedbackDrainHandle {
    pub fn wake(&self) {
        self.notify.notify_one();
    }
}

/// The process-wide drain handle, set by the first spawn. `feedback.submit`
/// wakes through [`wake_feedback_drain`] without holding a reference.
static DRAIN_HANDLE: OnceLock<FeedbackDrainHandle> = OnceLock::new();

/// Wake the spawned drain (a no-op when none is running — e.g. under tests
/// or an embedder that never spawned one). Called by `feedback.submit` after
/// each durable enqueue, so a submission starts draining without any poll.
pub fn wake_feedback_drain() {
    if let Some(handle) = DRAIN_HANDLE.get() {
        handle.wake();
    }
}

/// Spawn the drain for this daemon: derives the CAR state root the same way
/// the `feedback.*` surface does, builds the live Parslee transport, and
/// parks until the first `feedback.submit` (after one boot pass that uploads
/// anything spooled while the daemon was down). Call once at boot, like
/// `spawn_stale_registry_reaper`; the task dies with the runtime.
pub fn spawn_feedback_drain(state: &ServerState) {
    let car_home = match crate::feedback::car_home_dir(state) {
        Ok(home) => home,
        Err(e) => {
            tracing::warn!(target: "car::feedback", error = %e, "feedback drain not started");
            return;
        }
    };
    // LAZY on purpose: building the live transport reads the published auth
    // state through the secret store, and a cold daemon must boot with ZERO
    // credential reads (car#661). Construction happens on the first round
    // that actually has queued work.
    let transport: Arc<dyn DrainTransport> = Arc::new(LazyLiveTransport::new());
    let handle = spawn_feedback_drain_with(car_home, transport, DrainConfig::default());
    // First spawn wins; a second daemon-in-process (tests) keeps its own handle.
    let _ = DRAIN_HANDLE.set(handle);
}

/// Spawn a drain loop over an explicit spool root + transport + config — the
/// seam production wiring and tests share.
pub fn spawn_feedback_drain_with(
    car_home: PathBuf,
    transport: Arc<dyn DrainTransport>,
    config: DrainConfig,
) -> FeedbackDrainHandle {
    let notify = Arc::new(Notify::new());
    let handle = FeedbackDrainHandle {
        notify: notify.clone(),
    };
    let spool_root = car_home.join(FEEDBACK_OUTBOX_DIR);
    tokio::spawn(async move {
        let mut last_upload: Option<Instant> = None;
        // The F13 attempt ledger lives exactly as long as this loop: one
        // daemon lifetime, one budget per entry.
        let mut ledger = RetryLedger::default();
        loop {
            // Drain until the outbox is empty or everything holds.
            let mut backoff = config.initial_backoff;
            loop {
                let outcome = run_drain_round(
                    &spool_root,
                    transport.as_ref(),
                    &config,
                    &mut last_upload,
                    &mut ledger,
                )
                .await;
                match outcome {
                    RoundOutcome::Idle => break,
                    RoundOutcome::Progressed => {
                        backoff = config.initial_backoff;
                    }
                    RoundOutcome::AllHeld => {
                        // Re-check held entries later — or immediately on a
                        // new submit (which may also change the session).
                        wait_or_wake(&notify, config.held_recheck_interval).await;
                    }
                    RoundOutcome::Backoff { min_delay } => {
                        let delay = backoff.max(min_delay).min(config.max_backoff);
                        wait_or_wake(&notify, delay).await;
                        backoff = (backoff * 2).min(config.max_backoff);
                    }
                    RoundOutcome::RetryAfter(delay) => {
                        // Retry-After is honored (requirement 12) — up to the
                        // backoff ceiling: the header is advisory, and an
                        // unclamped value on this single sequential task would
                        // park every queued report until restart. The wait is
                        // interruptible like every other wait in this loop; a
                        // fresh submit's wake runs a round whose upload is
                        // still paced by `min_upload_interval`, and a repeat
                        // 429 simply re-arms this arm.
                        let delay = delay.min(config.max_backoff);
                        wait_or_wake(&notify, delay).await;
                    }
                }
            }
            // Outbox has no pending work: park until the next submit's Notify
            // — or until the LOCAL-ONLY park tick spots a PENDING entry that
            // arrived without one (a CLI direct enqueue, finding #11). The
            // tick reads the outbox and nothing else; it makes NO transport
            // call, so PRIV-1 holds while parked.
            loop {
                tokio::select! {
                    _ = notify.notified() => break,
                    _ = tokio::time::sleep(config.park_check_interval) => {
                        if spool_has_pending_entries(&spool_root) {
                            break;
                        }
                    }
                }
            }
        }
    });
    handle
}

/// LOCAL-ONLY park-tick probe: does the outbox hold a PENDING (Queued or
/// Sending) entry? Two-stage on purpose. The directory listing alone answers
/// the overwhelmingly common case — an empty outbox — without opening the
/// spool. Only when published entry directories exist does the probe ask the
/// spool's own list surface which of them are still pending: settled entries
/// (acknowledged/terminal) linger for the 14-day TTL and a stray hand-made
/// directory is not an entry, and neither may break the park (grace r2 — the
/// old any-directory answer made an idle-but-non-empty outbox run a spurious
/// drain round every tick). No transport is touched on any path. Entries
/// still in `.tmp-*` staging don't count; their publishing rename lands
/// before the enqueuer returns. An unreadable outbox reads as empty, the same
/// convention the round would report as a warning on the next wake.
fn spool_has_pending_entries(spool_root: &Path) -> bool {
    let Ok(dirents) = std::fs::read_dir(spool_root) else {
        // Missing/unreadable outbox dir == empty.
        return false;
    };
    let has_published_dir = dirents.flatten().any(|dirent| {
        !dirent.file_name().to_string_lossy().starts_with(".tmp-")
            && dirent.file_type().map(|t| t.is_dir()).unwrap_or(false)
    });
    if !has_published_dir {
        return false;
    }
    let Ok(spool) = Spool::open(spool_root) else {
        return false;
    };
    spool
        .list()
        .map(|rows| {
            rows.iter()
                .any(|row| matches!(row.state, SpoolState::Queued | SpoolState::Sending))
        })
        .unwrap_or(false)
}

async fn wait_or_wake(notify: &Notify, delay: Duration) {
    tokio::select! {
        _ = tokio::time::sleep(delay) => {}
        _ = notify.notified() => {}
    }
}

/// One pass over the outbox. Reads the spool BEFORE touching the transport:
/// an empty outbox returns [`RoundOutcome::Idle`] having made zero transport
/// calls (PRIV-1 / SPOOL-5). Every state change is applied to the durable
/// spool via its transition API. `ledger` is the loop-owned F13 attempt
/// ledger; a caller running rounds in isolation passes a fresh one.
pub async fn run_drain_round(
    spool_root: &Path,
    transport: &dyn DrainTransport,
    config: &DrainConfig,
    last_upload: &mut Option<Instant>,
    ledger: &mut RetryLedger,
) -> RoundOutcome {
    // All spool I/O is small bounded filesystem work; run_blocking is not
    // needed for correctness here, and the drain runs on its own task.
    let spool = match Spool::open(spool_root) {
        Ok(s) => s,
        Err(e) => {
            tracing::warn!(target: "car::feedback", error = %e, "feedback spool unavailable");
            return RoundOutcome::Backoff {
                min_delay: config.initial_backoff,
            };
        }
    };
    let entries = match spool.list() {
        Ok(rows) => rows,
        Err(e) => {
            tracing::warn!(target: "car::feedback", error = %e, "feedback spool list failed");
            return RoundOutcome::Backoff {
                min_delay: config.initial_backoff,
            };
        }
    };

    // Crash recovery (checklist #1's restart scenario): a Sending entry at
    // round start is a previous drain's in-flight leftover — hand it back to
    // Queued; re-sending under the same client_submission_id is IDEM-1-safe.
    let mut queued: Vec<_> = Vec::new();
    for entry in entries {
        match &entry.state {
            SpoolState::Queued => queued.push(entry),
            SpoolState::Sending => {
                if let Err(e) = spool.mark_queued(&entry.id) {
                    tracing::warn!(
                        target: "car::feedback",
                        entry = %entry.id, error = %e,
                        "stale Sending entry could not be recovered"
                    );
                } else {
                    queued.push(entry);
                }
            }
            SpoolState::Acknowledged { .. }
            | SpoolState::TerminalActionable { .. }
            | SpoolState::TerminalRejected { .. } => {}
        }
    }

    if queued.is_empty() {
        // PRIV-1: no queued work ⇒ the transport is never consulted.
        return RoundOutcome::Idle;
    }

    // Finding F13: drop ledger rows for entries no longer queued, then set
    // aside every entry past this process's retriable-attempt budget BEFORE
    // the eligibility probe — a capped entry stays Queued on disk (never
    // terminal, never pruned: requirement 13) and costs no network this
    // round. The cap-hit warning was logged once when the budget ran out;
    // each skip is debug-level only.
    ledger.retain_queued(&queued);
    let (queued, capped): (Vec<_>, Vec<_>) = queued
        .into_iter()
        .partition(|entry| !ledger.is_exhausted(&entry.id, config.max_retriable_attempts));
    for entry in &capped {
        tracing::debug!(
            target: "car::feedback",
            entry = %entry.id,
            "feedback entry past this daemon's retriable-attempt cap; skipped this round"
        );
    }
    if queued.is_empty() {
        // Only capped entries remain: held by this process, no transport
        // call — the held-recheck cadence keeps the loop calm.
        return RoundOutcome::AllHeld;
    }

    // Per-round eligibility cache: the verdict now comes from the anonymous
    // capability probe (`GET /api/v1/car-feedback/capability`) plus session
    // presence — both install-global, neither per-org — so it is keyed on the
    // LANE KIND only and N queued entries across any number of orgs cost at
    // most one probe per lane kind per round ("cache the probe briefly" —
    // the round IS the cache lifetime; the transport holds no cache).
    let mut eligibility: HashMap<&'static str, DrainEligibility> = HashMap::new();
    let mut progressed = false;
    let mut retriable: Option<Duration> = None;

    for entry in queued {
        // Anonymous intake is deliberately unavailable in v1. Do not spend a
        // bearer refresh + capability request rediscovering that fixed local
        // fact every held-recheck round.
        if matches!(entry.lane, IdentityLane::Anonymous) {
            tracing::debug!(
                target: "car::feedback",
                entry = %entry.id,
                "anonymous feedback entry held locally; v1 has no anonymous drain"
            );
            continue;
        }
        let cache_key = "authenticated";
        let verdict = match eligibility.get(&cache_key) {
            Some(v) => v.clone(),
            None => {
                let v = transport.drain_eligible(&entry.lane).await;
                eligibility.insert(cache_key, v.clone());
                v
            }
        };
        if let DrainEligibility::Hold { reason } = verdict {
            tracing::debug!(
                target: "car::feedback",
                entry = %entry.id, reason = %reason,
                "feedback entry held queued"
            );
            continue;
        }

        // SPOOL-7: pace uploads inside the shared org window.
        if let Some(last) = *last_upload {
            let since = last.elapsed();
            if since < config.min_upload_interval {
                tokio::time::sleep(config.min_upload_interval - since).await;
            }
        }

        if let Err(e) = spool.mark_sending(&entry.id) {
            tracing::warn!(
                target: "car::feedback",
                entry = %entry.id, error = %e,
                "mark_sending failed; skipping entry this round"
            );
            continue;
        }

        let bundle = match spool.load_bundle(&entry.id) {
            Ok(b) => b,
            Err(e) => {
                // A corrupt stored bundle can never upload — terminal with
                // the reason, not an infinite retry.
                let _ = spool.mark_terminal(
                    &entry.id,
                    TerminalReason::Rejected {
                        message: format!("stored bundle unreadable: {e}"),
                    },
                );
                progressed = true;
                continue;
            }
        };

        let attempt = transport
            .submit(&bundle, &entry.lane, &entry.client_submission_id)
            .await;
        *last_upload = Some(Instant::now());

        match attempt {
            Ok(SubmitOutcome { action, omitted }) => {
                report_omitted(&entry.id, &omitted);
                match action {
                    DrainAction::Acknowledge { server_id } => {
                        if apply(&spool, &entry.id, |s| {
                            s.mark_acknowledged(&entry.id, &server_id)
                        }) {
                            progressed = true;
                        }
                    }
                    DrainAction::Requeue { backoff } => {
                        apply(&spool, &entry.id, |s| s.mark_queued(&entry.id));
                        ledger.record_failure(&entry.id, config.max_retriable_attempts);
                        let delay = Duration::from_secs(backoff);
                        retriable = Some(retriable.map_or(delay, |d| d.max(delay)));
                    }
                    DrainAction::RequeueAfter { secs } => {
                        apply(&spool, &entry.id, |s| s.mark_queued(&entry.id));
                        // Rate-limited: stop the round — every further upload
                        // this round would burst the same window
                        // (requirement 12).
                        return RoundOutcome::RetryAfter(Duration::from_secs(secs));
                    }
                    DrainAction::TerminalActionable { reason } => {
                        let reason = match reason {
                            TransportActionableReason::AuthRequired => TerminalReason::AuthRequired,
                            TransportActionableReason::ReconsentRequired => {
                                TerminalReason::ReconsentRequired
                            }
                            TransportActionableReason::Forbidden => TerminalReason::Forbidden,
                        };
                        if apply(&spool, &entry.id, |s| s.mark_terminal(&entry.id, reason)) {
                            progressed = true;
                        }
                    }
                    DrainAction::TerminalRejected { message } => {
                        // The omitted notes ride the durable rejection
                        // message — the one settle path with a persisted
                        // free-text slot today.
                        let message = match omitted_suffix(&omitted) {
                            Some(suffix) => format!("{message}{suffix}"),
                            None => message,
                        };
                        if apply(&spool, &entry.id, |s| {
                            s.mark_terminal(&entry.id, TerminalReason::Rejected { message })
                        }) {
                            progressed = true;
                        }
                    }
                }
            }
            // Typed holds (anonymous intake not accepted, session lost between the
            // eligibility check and the send): back to Queued, no terminal.
            Err(FeedbackTransportError::AnonymousNotYetSupported)
            | Err(FeedbackTransportError::NoSession) => {
                apply(&spool, &entry.id, |s| s.mark_queued(&entry.id));
            }
            // Read-path taxonomy leaking into a submit is unexpected but must
            // not orphan the entry: a rejected bearer holds Queued exactly
            // like NoSession (the user may sign back in); a transport-level
            // fetch failure is retriable on the backoff schedule.
            Err(FeedbackTransportError::Unauthorized) => {
                apply(&spool, &entry.id, |s| s.mark_queued(&entry.id));
            }
            Err(FeedbackTransportError::FetchFailed(e)) => {
                tracing::warn!(
                    target: "car::feedback",
                    entry = %entry.id, error = %e,
                    "feedback submit transport failure; will retry"
                );
                apply(&spool, &entry.id, |s| s.mark_queued(&entry.id));
                ledger.record_failure(&entry.id, config.max_retriable_attempts);
                let delay = config.initial_backoff;
                retriable = Some(retriable.map_or(delay, |d| d.max(delay)));
            }
        }
    }

    if let Some(min_delay) = retriable {
        RoundOutcome::Backoff { min_delay }
    } else if progressed {
        RoundOutcome::Progressed
    } else {
        RoundOutcome::AllHeld
    }
}

/// One `"; omitted: a; b"` suffix for a settle message, `None` when the full
/// bundle rode the wire.
fn omitted_suffix(omitted: &[String]) -> Option<String> {
    if omitted.is_empty() {
        None
    } else {
        Some(format!(" (omitted: {})", omitted.join("; ")))
    }
}

/// Surface the transport's omitted-item notes (the 413/preflight fallback
/// ladder's honesty trail). For a rejected entry they are folded into the
/// durable message at the call site; for an acknowledged entry the spool has
/// no free-text slot yet, so they are logged — attaching them to the
/// acknowledged `state.json` awaits a spool note API (recorded as an open
/// item on the PR).
fn report_omitted(id: &SpoolEntryId, omitted: &[String]) {
    if let Some(suffix) = omitted_suffix(omitted) {
        tracing::warn!(
            target: "car::feedback",
            entry = %id,
            "feedback upload sent with omissions{suffix}"
        );
    }
}

/// Apply one spool transition, logging (never panicking) on failure — the
/// drain must survive a hand-damaged entry and keep serving the rest.
fn apply(
    spool: &Spool,
    id: &SpoolEntryId,
    transition: impl FnOnce(&Spool) -> std::io::Result<()>,
) -> bool {
    match transition(spool) {
        Ok(()) => true,
        Err(e) => {
            tracing::warn!(
                target: "car::feedback",
                entry = %id, error = %e,
                "spool transition failed"
            );
            false
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_feedback_core::bundle::{collect, CollectInputs, RedactedBundle};
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Mutex;
    use tempfile::TempDir;

    /// Counting, scriptable mock transport. Every call — eligibility included
    /// — counts as a transport call for the PRIV-1 distinguishing test.
    struct MockTransport {
        eligible_calls: AtomicUsize,
        submit_calls: AtomicUsize,
        eligibility: Mutex<DrainEligibility>,
        /// Successive submit results; the last one repeats.
        script: Mutex<Vec<Result<DrainAction, FeedbackTransportError>>>,
        /// Omitted-item notes attached to every scripted Ok outcome (the
        /// transport fallback-ladder honesty trail).
        omitted: Mutex<Vec<String>>,
        /// Virtual timestamps of each submit (paused-clock pacing asserts).
        submit_at: Mutex<Vec<Instant>>,
        submitted_ids: Mutex<Vec<String>>,
    }

    impl MockTransport {
        fn new(action: DrainAction) -> Self {
            MockTransport {
                eligible_calls: AtomicUsize::new(0),
                submit_calls: AtomicUsize::new(0),
                eligibility: Mutex::new(DrainEligibility::Eligible),
                script: Mutex::new(vec![Ok(action)]),
                omitted: Mutex::new(Vec::new()),
                submit_at: Mutex::new(Vec::new()),
                submitted_ids: Mutex::new(Vec::new()),
            }
        }

        fn holding(reason: &str) -> Self {
            let t = Self::new(DrainAction::Acknowledge {
                server_id: "unused".into(),
            });
            *t.eligibility.lock().unwrap() = DrainEligibility::Hold {
                reason: reason.to_string(),
            };
            t
        }

        fn total_calls(&self) -> usize {
            self.eligible_calls.load(Ordering::SeqCst) + self.submit_calls.load(Ordering::SeqCst)
        }
    }

    #[async_trait]
    impl DrainTransport for MockTransport {
        async fn drain_eligible(&self, _lane: &IdentityLane) -> DrainEligibility {
            self.eligible_calls.fetch_add(1, Ordering::SeqCst);
            self.eligibility.lock().unwrap().clone()
        }
        async fn submit(
            &self,
            _bundle: &RedactedBundle,
            _lane: &IdentityLane,
            client_submission_id: &str,
        ) -> Result<SubmitOutcome, FeedbackTransportError> {
            self.submit_calls.fetch_add(1, Ordering::SeqCst);
            self.submit_at.lock().unwrap().push(Instant::now());
            self.submitted_ids
                .lock()
                .unwrap()
                .push(client_submission_id.to_string());
            let mut script = self.script.lock().unwrap();
            let next = if script.len() > 1 {
                script.remove(0)
            } else {
                script[0].clone()
            };
            next.map(|action| SubmitOutcome {
                action,
                omitted: self.omitted.lock().unwrap().clone(),
            })
        }
    }

    fn bundle() -> RedactedBundle {
        let tmp = TempDir::new().unwrap();
        collect(CollectInputs {
            description: "the command deck window went blank".to_string(),
            state_root: Some(tmp.path().to_path_buf()),
            ..CollectInputs::default()
        })
        .unwrap()
    }

    fn auth_lane() -> IdentityLane {
        IdentityLane::Authenticated {
            org_id: "org_abc".to_string(),
        }
    }

    fn fast_config() -> DrainConfig {
        DrainConfig {
            min_upload_interval: Duration::from_secs(15),
            initial_backoff: Duration::from_secs(1),
            max_backoff: Duration::from_secs(8),
            held_recheck_interval: Duration::from_secs(60),
            park_check_interval: Duration::from_secs(60),
            max_retriable_attempts: DrainConfig::default().max_retriable_attempts,
        }
    }

    /// A fresh per-call ledger — the isolation most rounds below want (no
    /// cap can trip inside one round: an entry fails at most once per round).
    fn ledger() -> RetryLedger {
        RetryLedger::default()
    }

    fn enqueue(root: &Path, lane: IdentityLane) -> SpoolEntryId {
        let spool = Spool::open(root).unwrap();
        spool.enqueue(&bundle(), lane, "title").unwrap()
    }

    /// Re-open the spool FROM DISK and read one entry's persisted state —
    /// the durable-state trace (checklist #1), never an in-memory mirror.
    fn persisted_state(root: &Path, id: &SpoolEntryId) -> SpoolState {
        Spool::open(root)
            .unwrap()
            .list()
            .unwrap()
            .into_iter()
            .find(|e| &e.id == id)
            .expect("entry on disk")
            .state
    }

    // ---- PRIV-1 / SPOOL-5: the distinguishing test -------------------------

    #[tokio::test]
    async fn lazy_live_transport_construction_failure_holds_without_submitting() {
        let transport = LazyLiveTransport::failing("fixture construction failure");
        let verdict = transport.drain_eligible(&auth_lane()).await;
        assert_eq!(
            verdict,
            DrainEligibility::Hold {
                reason: "feedback transport unavailable: fixture construction failure".to_string()
            }
        );
    }

    #[tokio::test]
    async fn empty_outbox_produces_no_transport_calls() {
        // Distinguishing scenario: a ticker-style drain (like every other
        // daemon loop) would consult the transport on a schedule; this drain
        // must return Idle from an empty outbox having made ZERO transport
        // calls — not even an eligibility/entitlements probe.
        let tmp = TempDir::new().unwrap();
        let spool_root = tmp.path().join("feedback-outbox");
        let transport = MockTransport::new(DrainAction::Acknowledge {
            server_id: "never".into(),
        });
        let mut last = None;
        let outcome = run_drain_round(
            &spool_root,
            &transport,
            &fast_config(),
            &mut last,
            &mut ledger(),
        )
        .await;
        assert_eq!(outcome, RoundOutcome::Idle);
        assert_eq!(
            transport.total_calls(),
            0,
            "empty outbox must touch nothing"
        );
    }

    // ---- happy path: persisted queued → sending → acknowledged -------------

    #[tokio::test]
    async fn acknowledged_entry_persists_the_server_id_on_disk() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, auth_lane());
        let transport = MockTransport::new(DrainAction::Acknowledge {
            server_id: "row-7".into(),
        });
        let mut last = None;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::Progressed);
        assert_eq!(
            persisted_state(&root, &id),
            SpoolState::Acknowledged {
                server_id: "row-7".to_string()
            }
        );
        // The submit carried the entry's immutable idempotency key (IDEM-1).
        let sent = transport.submitted_ids.lock().unwrap().clone();
        assert_eq!(sent.len(), 1);
        assert!(!sent[0].is_empty());
    }

    // ---- requirement 3 / DEG-1/DEG-3: the capability gate -------------------

    #[tokio::test]
    async fn hold_verdict_keeps_entries_queued_then_capability_flip_drains_them() {
        // DEG-1: transport says Hold (the capability probe does not
        // advertise authenticated intake) → the spool holds, nothing is
        // submitted. DEG-3: the verdict flips (the capability appears) →
        // the SAME entry drains unchanged.
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, auth_lane());
        let transport =
            MockTransport::holding("capability does not advertise authenticated intake");
        let mut last = None;

        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::AllHeld);
        assert_eq!(transport.submit_calls.load(Ordering::SeqCst), 0);
        assert_eq!(persisted_state(&root, &id), SpoolState::Queued);

        *transport.eligibility.lock().unwrap() = DrainEligibility::Eligible;
        *transport.script.lock().unwrap() = vec![Ok(DrainAction::Acknowledge {
            server_id: "row-1".into(),
        })];
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::Progressed);
        assert_eq!(
            persisted_state(&root, &id),
            SpoolState::Acknowledged {
                server_id: "row-1".to_string()
            }
        );
    }

    #[tokio::test]
    async fn capability_probe_is_cached_per_round_across_orgs() {
        // The eligibility verdict is install-global (capability probe +
        // session), never per-org: two queued entries for two DIFFERENT orgs
        // must cost exactly ONE drain_eligible call in a round. The old
        // per-org cache key made this two calls — this test fails there.
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        enqueue(&root, auth_lane());
        enqueue(
            &root,
            IdentityLane::Authenticated {
                org_id: "org_other".to_string(),
            },
        );
        let transport = MockTransport::holding("capability unavailable");
        let mut last = None;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::AllHeld);
        assert_eq!(
            transport.eligible_calls.load(Ordering::SeqCst),
            1,
            "one capability-backed eligibility probe per lane kind per round"
        );
    }

    #[tokio::test]
    async fn anonymous_entries_hold_queued_without_a_submit() {
        // v1 ruling (a): the server does not accept anonymous intake
        // (`anonymousIntake: false` on the capability probe); DEG-1 covers
        // these entries.
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, IdentityLane::Anonymous);
        let transport = MockTransport::holding("server does not accept anonymous feedback");
        let mut last = None;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::AllHeld);
        assert_eq!(transport.submit_calls.load(Ordering::SeqCst), 0);
        assert_eq!(
            transport.eligible_calls.load(Ordering::SeqCst),
            0,
            "anonymous v1 entries must not trigger a capability probe"
        );
        assert_eq!(persisted_state(&root, &id), SpoolState::Queued);
    }

    // ---- SPOOL-6: taxonomy applied to durable state -------------------------

    #[tokio::test]
    async fn corrupt_bundle_settles_terminal_instead_of_retrying_forever() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, auth_lane());
        std::fs::write(root.join(id.as_str()).join("bundle.json"), b"{corrupt").unwrap();
        let transport = MockTransport::new(DrainAction::Acknowledge {
            server_id: "must-not-submit".into(),
        });
        let mut last = None;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::Progressed);
        assert_eq!(transport.submit_calls.load(Ordering::SeqCst), 0);
        match persisted_state(&root, &id) {
            SpoolState::TerminalRejected { message } => {
                assert!(message.contains("stored bundle unreadable"), "{message}");
            }
            state => panic!("corrupt bundle must settle terminal, got {state:?}"),
        }
    }

    #[tokio::test]
    async fn submit_unauthorized_returns_entry_to_queued_hold() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, auth_lane());
        let transport = MockTransport::new(DrainAction::Acknowledge {
            server_id: "unused".into(),
        });
        *transport.script.lock().unwrap() = vec![Err(FeedbackTransportError::Unauthorized)];
        let mut last = None;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::AllHeld);
        assert_eq!(persisted_state(&root, &id), SpoolState::Queued);
    }

    #[tokio::test]
    async fn terminal_actionable_and_rejected_persist_and_never_retry() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let auth_id = enqueue(&root, auth_lane());
        let transport = MockTransport::new(DrainAction::TerminalActionable {
            reason: TransportActionableReason::ReconsentRequired,
        });
        let mut last = None;
        run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert!(matches!(
            persisted_state(&root, &auth_id),
            SpoolState::TerminalActionable {
                reason: car_feedback_core::spool::ActionableReason::ReconsentRequired
            }
        ));

        // SPOOL-6's "never retried": another round makes no further submit.
        let before = transport.submit_calls.load(Ordering::SeqCst);
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::Idle);
        assert_eq!(transport.submit_calls.load(Ordering::SeqCst), before);

        // 400-class: terminal with the server's message, persisted.
        // (Reset pacing so this unpaused test doesn't sleep a real interval.)
        last = None;
        let rejected_id = enqueue(&root, auth_lane());
        *transport.script.lock().unwrap() = vec![Ok(DrainAction::TerminalRejected {
            message: "HTTP 400: description invalid".into(),
        })];
        run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(
            persisted_state(&root, &rejected_id),
            SpoolState::TerminalRejected {
                message: "HTTP 400: description invalid".to_string()
            }
        );
    }

    #[tokio::test]
    async fn retriable_failure_requeues_durably_and_reports_backoff() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, auth_lane());
        let transport = MockTransport::new(DrainAction::Requeue { backoff: 30 });
        let mut last = None;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(
            outcome,
            RoundOutcome::Backoff {
                min_delay: Duration::from_secs(30)
            }
        );
        assert_eq!(persisted_state(&root, &id), SpoolState::Queued);
    }

    #[tokio::test(start_paused = true)]
    async fn retry_after_stops_the_round_and_is_honored_before_the_next_upload() {
        // SPOOL-7 against a mock 429: the round stops immediately (no burst
        // into the same window) and the next submit happens no sooner than
        // Retry-After.
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let first = enqueue(&root, auth_lane());
        let second = enqueue(&root, auth_lane());
        let transport = MockTransport::new(DrainAction::Acknowledge {
            server_id: "row".into(),
        });
        *transport.script.lock().unwrap() = vec![
            Ok(DrainAction::RequeueAfter { secs: 40 }),
            Ok(DrainAction::Acknowledge {
                server_id: "row-a".into(),
            }),
            Ok(DrainAction::Acknowledge {
                server_id: "row-b".into(),
            }),
        ];
        let mut last = None;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::RetryAfter(Duration::from_secs(40)));
        // Only ONE submit happened — the 429 stopped the round before the
        // second entry could burst the same window.
        assert_eq!(transport.submit_calls.load(Ordering::SeqCst), 1);
        assert_eq!(persisted_state(&root, &first), SpoolState::Queued);
        assert_eq!(persisted_state(&root, &second), SpoolState::Queued);

        // Honor it, then drain: the next round's uploads are ≥40s later.
        tokio::time::sleep(Duration::from_secs(40)).await;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::Progressed);
        let stamps = transport.submit_at.lock().unwrap().clone();
        assert!(stamps.len() >= 2);
        assert!(
            stamps[1].duration_since(stamps[0]) >= Duration::from_secs(40),
            "second upload ran {:?} after the 429 — Retry-After not honored",
            stamps[1].duration_since(stamps[0])
        );
    }

    #[tokio::test(start_paused = true)]
    async fn uploads_pace_at_most_one_per_min_interval() {
        // SPOOL-7: two queued entries drain ≥15s apart (virtual clock).
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        enqueue(&root, auth_lane());
        enqueue(&root, auth_lane());
        let transport = MockTransport::new(DrainAction::Acknowledge {
            server_id: "row".into(),
        });
        let mut last = None;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::Progressed);
        let stamps = transport.submit_at.lock().unwrap().clone();
        assert_eq!(stamps.len(), 2);
        assert!(
            stamps[1].duration_since(stamps[0]) >= Duration::from_secs(15),
            "uploads {:?} apart — pacing not applied",
            stamps[1].duration_since(stamps[0])
        );
    }

    // ---- crash recovery (checklist #1's restart scenario) -------------------

    #[tokio::test]
    async fn stale_sending_entry_from_a_crashed_drain_recovers_and_resends_same_id() {
        // A daemon killed mid-upload leaves the entry Sending on disk. The
        // next round recovers it to Queued and re-sends it under the SAME
        // client_submission_id (IDEM-1 makes the replay safe server-side).
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, auth_lane());
        let spool = Spool::open(&root).unwrap();
        spool.mark_sending(&id).unwrap();
        let original_csid = spool
            .list()
            .unwrap()
            .into_iter()
            .find(|e| e.id == id)
            .unwrap()
            .client_submission_id;
        drop(spool); // "crash"

        let transport = MockTransport::new(DrainAction::Acknowledge {
            server_id: "row-1".into(),
        });
        let mut last = None;
        let outcome =
            run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        assert_eq!(outcome, RoundOutcome::Progressed);
        assert_eq!(
            persisted_state(&root, &id),
            SpoolState::Acknowledged {
                server_id: "row-1".to_string()
            }
        );
        assert_eq!(
            transport.submitted_ids.lock().unwrap().as_slice(),
            &[original_csid],
            "the recovered entry must re-send its ORIGINAL idempotency key"
        );
    }

    // ---- the spawned loop wakes on submit, parks when idle ------------------

    #[tokio::test(start_paused = true)]
    async fn spawned_drain_parks_idle_and_drains_on_wake() {
        let tmp = TempDir::new().unwrap();
        let car_home = tmp.path().to_path_buf();
        let root = car_home.join(FEEDBACK_OUTBOX_DIR);
        let transport = Arc::new(MockTransport::new(DrainAction::Acknowledge {
            server_id: "row-1".into(),
        }));
        let handle = spawn_feedback_drain_with(car_home, transport.clone(), fast_config());

        // PARK-PHASE assertion (PRIV-1's distinguishing half): a long virtual
        // idle spans dozens of 60s park ticks — each tick may read the disk,
        // but with an EMPTY outbox not one transport call (not even an
        // eligibility probe) may happen.
        tokio::time::sleep(Duration::from_secs(3600)).await;
        assert_eq!(
            transport.total_calls(),
            0,
            "an empty-outbox park (incl. its local disk ticks) must never touch the transport"
        );

        // Enqueue + wake (what feedback.submit does) → the entry drains.
        let id = enqueue(&root, auth_lane());
        handle.wake();
        // Let the loop run; paused clock auto-advances through its sleeps.
        for _ in 0..200 {
            tokio::task::yield_now().await;
            if transport.submit_calls.load(Ordering::SeqCst) > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert_eq!(transport.submit_calls.load(Ordering::SeqCst), 1);
        assert_eq!(
            persisted_state(&root, &id),
            SpoolState::Acknowledged {
                server_id: "row-1".to_string()
            }
        );
    }

    // ---- finding #11: the park tick catches CLI direct enqueues -------------

    #[tokio::test(start_paused = true)]
    async fn cli_direct_enqueue_while_parked_drains_within_one_tick() {
        // Distinguishing scenario: the CLI writes straight to the spool while
        // the daemon idles — NOTHING calls the drain handle. A Notify-only
        // drain parks forever; the park tick's local disk check must pick the
        // entry up within one park_check_interval.
        let tmp = TempDir::new().unwrap();
        let car_home = tmp.path().to_path_buf();
        let root = car_home.join(FEEDBACK_OUTBOX_DIR);
        let transport = Arc::new(MockTransport::new(DrainAction::Acknowledge {
            server_id: "row-cli".into(),
        }));
        let _handle = spawn_feedback_drain_with(car_home, transport.clone(), fast_config());

        // Let the loop reach its park (boot round on an empty outbox).
        tokio::time::sleep(Duration::from_secs(1)).await;
        assert_eq!(transport.total_calls(), 0);

        // The "CLI": a second Spool handle, no RPC, no wake().
        let id = enqueue(&root, auth_lane());

        // Within one park tick (60s here) the entry must drain.
        for _ in 0..200 {
            tokio::task::yield_now().await;
            if transport.submit_calls.load(Ordering::SeqCst) > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
        assert_eq!(
            transport.submit_calls.load(Ordering::SeqCst),
            1,
            "a parked drain must catch a direct spool enqueue via the local tick"
        );
        assert_eq!(
            persisted_state(&root, &id),
            SpoolState::Acknowledged {
                server_id: "row-cli".to_string()
            }
        );
    }

    /// Distinguishing test (grace r2, park probe): the old probe answered
    /// "any directory", so a settled entry waiting out its 14-day TTL — or one
    /// stray hand-made directory — broke the park every tick forever. The
    /// probe must answer PENDING entries only: false for absence, staging,
    /// files, stray dirs, and settled entries; true for Queued and Sending.
    #[test]
    fn park_probe_sees_only_pending_entries() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join(FEEDBACK_OUTBOX_DIR);
        // Absent dir: empty (and no transport implications at all).
        assert!(!spool_has_pending_entries(&root));
        std::fs::create_dir_all(root.join(".tmp-half-written")).unwrap();
        std::fs::write(root.join("stray-file"), b"x").unwrap();
        assert!(
            !spool_has_pending_entries(&root),
            "staging dirs and files don't count"
        );
        // A stray directory that is not a spool entry (no state.json).
        std::fs::create_dir_all(root.join("00000000000000000000-not-an-entry")).unwrap();
        assert!(
            !spool_has_pending_entries(&root),
            "a stray directory is not a pending entry"
        );
        // A settled entry — acknowledged, lingering for the TTL.
        let settled = enqueue(&root, auth_lane());
        {
            let spool = Spool::open(&root).unwrap();
            spool.mark_sending(&settled).unwrap();
            spool.mark_acknowledged(&settled, "row-1").unwrap();
        }
        assert_eq!(
            persisted_state(&root, &settled),
            SpoolState::Acknowledged {
                server_id: "row-1".to_string()
            }
        );
        assert!(
            !spool_has_pending_entries(&root),
            "a settled entry must not break the park"
        );
        // A Queued entry: pending.
        let queued = enqueue(&root, auth_lane());
        assert!(spool_has_pending_entries(&root));
        // A Sending leftover (crashed drain): pending too — the round recovers it.
        Spool::open(&root).unwrap().mark_sending(&queued).unwrap();
        assert!(spool_has_pending_entries(&root));
    }

    // ---- grace r2: Retry-After is bounded and interruptible -----------------

    /// Distinguishing test: the loop used to `sleep(delay)` the classified
    /// Retry-After verbatim — a `999999999` header parked the single drain
    /// task for ~31 years (this test's virtual minute would find the entry
    /// still Queued after one submit). Clamped to `max_backoff` (8s here),
    /// the entry retries and settles within the minute.
    #[tokio::test(start_paused = true)]
    async fn absurd_retry_after_is_clamped_to_the_backoff_ceiling() {
        let tmp = TempDir::new().unwrap();
        let car_home = tmp.path().to_path_buf();
        let root = car_home.join(FEEDBACK_OUTBOX_DIR);
        let transport = Arc::new(MockTransport::new(DrainAction::Acknowledge {
            server_id: "row-1".into(),
        }));
        *transport.script.lock().unwrap() = vec![
            Ok(DrainAction::RequeueAfter { secs: 999_999_999 }),
            Ok(DrainAction::Acknowledge {
                server_id: "row-1".into(),
            }),
        ];
        let handle = spawn_feedback_drain_with(car_home, transport.clone(), fast_config());
        tokio::time::sleep(Duration::from_secs(1)).await;
        let id = enqueue(&root, auth_lane());
        handle.wake();
        // Let the first attempt (the 429) land.
        for _ in 0..200 {
            tokio::task::yield_now().await;
            if transport.submit_calls.load(Ordering::SeqCst) >= 1 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert_eq!(transport.submit_calls.load(Ordering::SeqCst), 1);
        assert_eq!(persisted_state(&root, &id), SpoolState::Queued);

        // One virtual minute covers the clamped wait (≤ max_backoff = 8s)
        // plus the 15s upload pacing; the unclamped loop is still asleep.
        tokio::time::sleep(Duration::from_secs(60)).await;
        assert_eq!(
            transport.submit_calls.load(Ordering::SeqCst),
            2,
            "the drain must retry within the backoff ceiling, not the header's decades"
        );
        assert_eq!(
            persisted_state(&root, &id),
            SpoolState::Acknowledged {
                server_id: "row-1".to_string()
            }
        );
    }

    /// Distinguishing test: the Retry-After wait was a plain sleep, so a
    /// submit during it changed nothing until the delay ran out. Routed
    /// through `wait_or_wake`, the wake ends the wait; the next round's upload
    /// is still spaced by `min_upload_interval` (15s), so by t+30s the retry
    /// has happened — with the old non-interruptible 300s sleep it has not.
    #[tokio::test(start_paused = true)]
    async fn submit_wake_ends_a_retry_after_wait_early() {
        let tmp = TempDir::new().unwrap();
        let car_home = tmp.path().to_path_buf();
        let root = car_home.join(FEEDBACK_OUTBOX_DIR);
        let transport = Arc::new(MockTransport::new(DrainAction::Acknowledge {
            server_id: "row".into(),
        }));
        *transport.script.lock().unwrap() = vec![
            Ok(DrainAction::RequeueAfter { secs: 300 }),
            Ok(DrainAction::Acknowledge {
                server_id: "row-a".into(),
            }),
            Ok(DrainAction::Acknowledge {
                server_id: "row-b".into(),
            }),
        ];
        // A ceiling ABOVE the header, so the clamp alone cannot explain an
        // early retry — only the wake can.
        let mut config = fast_config();
        config.max_backoff = Duration::from_secs(600);
        let handle = spawn_feedback_drain_with(car_home, transport.clone(), config);
        tokio::time::sleep(Duration::from_secs(1)).await;
        let first = enqueue(&root, auth_lane());
        handle.wake();
        for _ in 0..200 {
            tokio::task::yield_now().await;
            if transport.submit_calls.load(Ordering::SeqCst) >= 1 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        assert_eq!(
            transport.submit_calls.load(Ordering::SeqCst),
            1,
            "the 429 landed"
        );
        assert_eq!(persisted_state(&root, &first), SpoolState::Queued);

        // The user submits again while the drain is waiting out the 429.
        let second = enqueue(&root, auth_lane());
        handle.wake();
        tokio::time::sleep(Duration::from_secs(30)).await;
        assert!(
            transport.submit_calls.load(Ordering::SeqCst) >= 2,
            "a submit wake must end the Retry-After wait (calls: {})",
            transport.submit_calls.load(Ordering::SeqCst)
        );
        assert_eq!(
            persisted_state(&root, &first),
            SpoolState::Acknowledged {
                server_id: "row-a".to_string()
            },
            "the oldest queued entry is retried first"
        );
        // Let the second entry drain too — pacing spaces it 15s after the first.
        tokio::time::sleep(Duration::from_secs(30)).await;
        assert_eq!(
            persisted_state(&root, &second),
            SpoolState::Acknowledged {
                server_id: "row-b".to_string()
            }
        );
    }

    // ---- transport omitted-notes surface on settle (SubmitOutcome seam) -----

    #[tokio::test]
    async fn rejected_upload_with_omissions_persists_the_omitted_note() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, auth_lane());
        let transport = MockTransport::new(DrainAction::TerminalRejected {
            message: "HTTP 400: description invalid".into(),
        });
        *transport.omitted.lock().unwrap() = vec!["screenshot dropped (413 fallback)".to_string()];
        let mut last = None;
        run_drain_round(&root, &transport, &fast_config(), &mut last, &mut ledger()).await;
        match persisted_state(&root, &id) {
            SpoolState::TerminalRejected { message } => {
                assert!(
                    message.contains("screenshot dropped (413 fallback)"),
                    "the omitted note must persist in the durable rejection message: {message}"
                );
                assert!(message.contains("HTTP 400"));
            }
            other => panic!("expected TerminalRejected, got {other:?}"),
        }
    }

    // ---- finding F13: the per-process retriable-attempt cap ----------------

    /// Distinguishing test: without a cap, a server that keeps answering 5xx
    /// (or a network that keeps dropping the POST) gets one attempt per round
    /// forever. With the cap the entry is attempted exactly
    /// `max_retriable_attempts` times — both retriable arms count — then every
    /// further round in this process skips it: no submit, not even an
    /// eligibility probe, while it stays `Queued` on disk (never terminal,
    /// never pruned — requirement 13). A fresh ledger (a daemon restart) tries
    /// again.
    #[tokio::test(start_paused = true)]
    async fn retriable_failures_stop_after_the_per_process_cap_and_the_entry_stays_queued() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, auth_lane());
        let transport = MockTransport::new(DrainAction::Requeue { backoff: 1 });
        *transport.script.lock().unwrap() = vec![
            Ok(DrainAction::Requeue { backoff: 1 }),
            Err(FeedbackTransportError::FetchFailed(
                "connection reset".into(),
            )),
            Ok(DrainAction::Requeue { backoff: 1 }),
        ];
        let mut config = fast_config();
        config.max_retriable_attempts = 3;
        let mut last = None;
        let mut ledger = RetryLedger::default();

        for round in 1..=3 {
            let outcome = run_drain_round(&root, &transport, &config, &mut last, &mut ledger).await;
            assert!(
                matches!(outcome, RoundOutcome::Backoff { .. }),
                "round {round}: {outcome:?}"
            );
            assert_eq!(transport.submit_calls.load(Ordering::SeqCst), round);
            assert_eq!(persisted_state(&root, &id), SpoolState::Queued);
        }

        // Budget spent: further rounds skip it without touching the transport.
        let probes_before = transport.eligible_calls.load(Ordering::SeqCst);
        for _ in 0..3 {
            let outcome = run_drain_round(&root, &transport, &config, &mut last, &mut ledger).await;
            assert_eq!(outcome, RoundOutcome::AllHeld);
        }
        assert_eq!(
            transport.submit_calls.load(Ordering::SeqCst),
            3,
            "no submit past the cap"
        );
        assert_eq!(
            transport.eligible_calls.load(Ordering::SeqCst),
            probes_before,
            "a capped entry costs no network — not even the eligibility probe"
        );
        assert_eq!(
            persisted_state(&root, &id),
            SpoolState::Queued,
            "capped ⇒ still Queued on disk: never a terminal state, never pruned"
        );

        // A daemon restart == a fresh ledger: the entry gets a new budget.
        let mut restarted = RetryLedger::default();
        let outcome = run_drain_round(&root, &transport, &config, &mut last, &mut restarted).await;
        assert!(matches!(outcome, RoundOutcome::Backoff { .. }));
        assert_eq!(transport.submit_calls.load(Ordering::SeqCst), 4);
    }

    /// 429 is pacing, not a failure of the entry: `Retry-After` rounds never
    /// consume the F13 budget — with a cap of ONE, two consecutive 429s still
    /// leave the entry sendable and the third round acknowledges it. (Were
    /// 429 counted, round two would already skip it as AllHeld.)
    #[tokio::test(start_paused = true)]
    async fn retry_after_does_not_consume_the_attempt_budget() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("feedback-outbox");
        let id = enqueue(&root, auth_lane());
        let transport = MockTransport::new(DrainAction::Acknowledge {
            server_id: "row".into(),
        });
        *transport.script.lock().unwrap() = vec![
            Ok(DrainAction::RequeueAfter { secs: 1 }),
            Ok(DrainAction::RequeueAfter { secs: 1 }),
            Ok(DrainAction::Acknowledge {
                server_id: "row-1".into(),
            }),
        ];
        let mut config = fast_config();
        config.max_retriable_attempts = 1;
        let mut last = None;
        let mut ledger = RetryLedger::default();

        for _ in 0..2 {
            let outcome = run_drain_round(&root, &transport, &config, &mut last, &mut ledger).await;
            assert_eq!(outcome, RoundOutcome::RetryAfter(Duration::from_secs(1)));
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
        let outcome = run_drain_round(&root, &transport, &config, &mut last, &mut ledger).await;
        assert_eq!(outcome, RoundOutcome::Progressed);
        assert_eq!(
            persisted_state(&root, &id),
            SpoolState::Acknowledged {
                server_id: "row-1".to_string()
            }
        );
    }
}