openlatch-client 0.1.18

//! Cloud forwarding worker — long-lived background task that POSTs hook events
//! to the OpenLatch cloud API.
//!
//! Architecture (D-01 through D-05):
//! - Single `tokio::spawn` task owns the `mpsc::Receiver<CloudEvent>`
//! - Events accumulate in a buffer and flush as one multi-event CloudEvents
//!   batch on whichever comes first: `batch_max_events` reached, or
//!   `batch_max_wait_ms` elapsed since the FIRST buffered event (D43/D45)
//! - Daemon handler uses `try_send()` only — zero blocking on the verdict path (CLOUD-09)
//! - Credential hot-reload via 60-second polling (D-07)
//! - Auth error state persisted to `cloud_state.json` for cross-process visibility (CLOUD-08)
//!
//! Error handling:
//! - 401/403: set auth_error flag, skip POSTs until credential refresh (D-14)
//! - 429: parse Retry-After, sleep, retry once (D-13)
//! - 5xx: sleep 2s, retry once, drop on second failure (D-12)
//! - Network: sleep 2s, retry once (D-12)
//! - Other 4xx: log, drop event (no retry)

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

use chrono::Utc;

use secrecy::SecretString;
use tokio::sync::mpsc;

use super::outbox::{DrainLimits, DrainOutcome, Outbox};
use super::{CloudConfig, CloudError, CloudEvent, CloudState, CredentialProvider};
use crate::core::cloud::envelope::build_cloud_headers;

/// Hard platform cap on the serialized size of one ingest request body.
///
/// `POST /api/v1/events/ingest` enforces this itself via
/// `require_body_under_256kb`, which reads `Content-Length` *before* parsing
/// the body — so an oversized batch fails with 413 identically on every
/// retry. Capping on event count alone would turn a burst of large `data`
/// payloads into a permanent stall with no useful diagnostic.
const MAX_BATCH_BYTES: usize = 262_144;

/// Hard platform cap on the number of events in one ingest request.
/// `batch_max_events` is the tuning knob; this is non-negotiable.
const MAX_BATCH_EVENTS: usize = 100;

/// How long the shutdown flush may run before the buffer is spooled instead.
///
/// Deliberately shorter than the daemon's own shutdown budget: the daemon drops
/// this task's `JoinHandle` after its window and the runtime then aborts the
/// task mid-await, so the flush must reach a terminal state — delivered or
/// spooled — strictly inside that window or the events are lost.
const SHUTDOWN_FLUSH_BUDGET: Duration = Duration::from_secs(3);

/// A deadline far enough out that the batch-flush timer is inert.
///
/// Deliberately a large *finite* offset rather than `Instant::MAX`: tokio's
/// timer wheel rejects instants beyond its maximum representable duration and
/// would panic instead of sleeping forever. The branch is disabled by its
/// `if deadline.is_some()` guard anyway — this only has to be a valid instant.
fn far_future() -> tokio::time::Instant {
    tokio::time::Instant::now() + Duration::from_secs(86_400)
}

// Credential hot-reload interval is a runtime parameter on CloudConfig
// (`credential_poll_interval_ms`) so tests can exercise the reload path
// without sleeping 60s.

/// Background cloud health probe interval.
///
/// Production builds honor `CloudState::next_health_delay` — 60s while
/// healthy, exponential backoff 5s → 30s while degraded. Integration tests
/// hard-code a 50ms interval so `record_health_ok` recovery paths can be
/// exercised without a real-time wait.
#[cfg(not(test))]
fn health_delay(state: &CloudState) -> Duration {
    state.next_health_delay()
}
#[cfg(test)]
fn health_delay(_state: &CloudState) -> Duration {
    Duration::from_millis(50)
}

/// Emit a single recovery info log when the worker transitions out of a
/// degraded streak, then clear the latch. No-op when not degraded.
fn note_recovery(degraded: &mut Option<&'static str>, state: &CloudState) {
    if let Some(reason) = degraded.take() {
        tracing::info!(
            reason,
            spooled = state.consecutive_drops(),
            "cloud worker: connectivity restored"
        );
    }
}

/// Shared auth_error clear path used by both the credential-refresh branch
/// and the out-of-band-clear branch in the credential-poll loop. Mirrors
/// the cleared value into the cross-process `cloud_state.json` and logs at
/// debug level if the persist fails (we already have the in-memory state).
fn clear_auth_error_latch(
    local: &mut bool,
    cloud_state: &CloudState,
    openlatch_dir: &Path,
    reason: &'static str,
) {
    *local = false;
    cloud_state.clear_auth_error();
    if let Err(e) = persist_cloud_state(openlatch_dir, false) {
        tracing::warn!(
            error = %e,
            reason,
            "cloud worker: failed to persist cloud_state.json on auth clear"
        );
    }
}

/// Threshold of consecutive live-event drops required before the detector
/// engages emergency mode. Tuned to avoid false-positives on small bursts:
/// a single full-batch (100 events) plus a few stragglers is fine, but a
/// sustained 100+ streak signals real congestion.
const EMERGENCY_DROP_THRESHOLD: u64 = 100;
/// Minimum duration the live-drop streak must last before engaging
/// emergency mode. Keeps a fast spike from latching the detector — only
/// sustained congestion (>10 s of dropping with the cap pegged) trips it.
const EMERGENCY_WINDOW_MS: u64 = 10_000;
/// Consecutive health ticks with zero live drops required to clear
/// emergency mode. Two ticks debounces a brief lull mid-streak.
const EMERGENCY_RECOVERY_TICKS: u32 = 2;

/// Channel-depth (as a percentage of `channel_size`) at which the
/// high-water detector starts a pressure window. Sits above the replay
/// producer's 75 % yield gate so the detector only fires when yield-pacing
/// alone has failed to relieve pressure.
pub(crate) const HIGH_WATER_TRIP_PCT: u64 = 80;
/// Hysteresis floor — the window is cleared only after depth dips below
/// this. Sustained pressure between the trip and clear thresholds keeps
/// the streak alive.
pub(crate) const HIGH_WATER_CLEAR_PCT: u64 = 50;
/// Minimum duration the channel must sit at or above the trip threshold
/// before emergency mode engages on the high-water path. Mirrors
/// `EMERGENCY_WINDOW_MS` so a brief spike doesn't latch the detector.
pub(crate) const HIGH_WATER_WINDOW_MS: u64 = 10_000;

/// Per-event attempt cap inside a single daemon lifetime before an
/// outbox entry is quarantined. State is volatile (lost on restart) — if
/// the entry is still poison after a restart, it quarantines again in
/// another `OUTBOX_MAX_ATTEMPTS` passes. Five matches the typical retry
/// budget for permanently-rejecting 5xx envelopes without flushing
/// legitimately-transient failures too aggressively.
const OUTBOX_MAX_ATTEMPTS: u32 = 5;

/// Label fed to `cloud_channel_overflow_emergency.trigger`. Keeps the
/// PostHog dimension typo-proof on a closed enum.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum EmergencyTrigger {
    LiveDrops,
    ChannelHighWater,
}

impl EmergencyTrigger {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            Self::LiveDrops => "live_drops",
            Self::ChannelHighWater => "channel_high_water",
        }
    }
}

/// Direction of an emergency-mode transition. Keeps telemetry labels
/// typo-proof — the constructor takes `as_str` rather than a raw string.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum EmergencyTransition {
    Enter,
    Exit,
}

impl EmergencyTransition {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            Self::Enter => "enter",
            Self::Exit => "exit",
        }
    }
}

/// Detector tick: engage on sustained streak (drops OR high-water pressure),
/// clear after recovery debounce when BOTH signals are quiet.
fn update_emergency_mode(state: &CloudState, config: &CloudConfig, recovery_ticks: &mut u32) {
    let drops = state.consecutive_live_drops();
    let window_start = state.live_drops_window_start_ms();
    let window_duration = super::now_unix_ms().saturating_sub(window_start);
    let high_water_window_ms = state.channel_high_water_window_ms();
    let channel_size = config.channel_size;

    if !state.is_emergency_mode() {
        let live_drop_sustained = drops > EMERGENCY_DROP_THRESHOLD
            && window_start > 0
            && window_duration > EMERGENCY_WINDOW_MS;
        let high_water_sustained = high_water_window_ms > HIGH_WATER_WINDOW_MS;
        if !live_drop_sustained && !high_water_sustained {
            return;
        }
        state.set_emergency_mode(true);
        *recovery_ticks = 0;
        // Live drops dominate when both fire — they represent actual lost
        // events, the more urgent signal.
        let (trigger, reported_duration) = if live_drop_sustained {
            (EmergencyTrigger::LiveDrops, window_duration)
        } else {
            (EmergencyTrigger::ChannelHighWater, high_water_window_ms)
        };
        tracing::warn!(
            code = crate::error::ERR_CLOUD_CHANNEL_EMERGENCY,
            trigger = trigger.as_str(),
            drops_in_window = drops,
            window_duration_ms = reported_duration,
            channel_size,
            "cloud channel under emergency drop — pausing replay until backlog clears"
        );
        crate::telemetry::capture_global(
            crate::telemetry::Event::cloud_channel_overflow_emergency(
                EmergencyTransition::Enter.as_str(),
                trigger.as_str(),
                drops,
                reported_duration,
                channel_size,
                channel_size,
            ),
        );
        return;
    }

    // In emergency mode: clear only when BOTH signals are quiet for the
    // debounce window. A clean tick requires zero live drops AND no active
    // high-water streak so we don't re-engage immediately after clearing.
    if drops > 0 || high_water_window_ms > 0 {
        *recovery_ticks = 0;
        return;
    }
    *recovery_ticks = recovery_ticks.saturating_add(1);
    if *recovery_ticks < EMERGENCY_RECOVERY_TICKS {
        return;
    }
    state.set_emergency_mode(false);
    *recovery_ticks = 0;
    tracing::info!("cloud channel emergency recovered — resuming replay");
    crate::telemetry::capture_global(crate::telemetry::Event::cloud_channel_overflow_emergency(
        EmergencyTransition::Exit.as_str(),
        EmergencyTrigger::LiveDrops.as_str(),
        0,
        window_duration,
        0,
        channel_size,
    ));
}

/// Build the reqwest HTTP client used by the cloud worker.
///
/// Per D-04: owned by the worker, not shared with the daemon.
/// - `connect_timeout`: TCP connect timeout from config
/// - `timeout`: total request timeout from config
/// - `pool_max_idle_per_host(4)`: connection pooling (CLOUD-06)
/// - `use_rustls_tls()`: rustls-only TLS (security constraint: no OpenSSL)
///
/// # Panics
///
/// Panics if the client cannot be constructed (only possible with invalid configs).
pub fn build_cloud_client(config: &CloudConfig) -> reqwest::Client {
    reqwest::Client::builder()
        .connect_timeout(Duration::from_millis(config.timeout_connect_ms))
        .timeout(Duration::from_millis(config.timeout_total_ms))
        .pool_max_idle_per_host(4)
        .use_rustls_tls()
        .build()
        .expect("failed to build cloud reqwest client")
}

/// Main cloud worker loop.
///
/// Consumes events from `rx`, accumulates them into a batch, and POSTs the
/// batch to the cloud API. Handles all error states inline: auth errors, rate
/// limiting, server errors, network errors.
///
/// Exits when the channel is closed (all senders dropped at daemon shutdown)
/// or when `shutdown` fires, flushing whatever is buffered either way.
///
/// # Parameters
///
/// - `rx`: receiver end of the bounded mpsc channel
/// - `credential_provider`: source of the current API key (hot-reloaded every 60s)
/// - `config`: cloud forwarding configuration
/// - `cloud_state`: shared in-process auth error flag (readable by AppState/status command)
/// - `openlatch_dir`: base directory for writing `cloud_state.json`
/// - `outbox`: durable spool for events that survive their retry, if enabled
/// - `shutdown`: explicit shutdown signal. Channel closure alone is **not** a
///   reliable trigger — `cloud_tx` lives on the `Arc<AppState>` and is cloned
///   into the config monitor and the tamper reconciler, so the mpsc may not
///   close promptly (or at all) on `openlatch stop`. `None` means "no
///   explicit signal", used by tests that shut down by dropping the sender.
pub async fn run_cloud_worker(
    rx: mpsc::Receiver<CloudEvent>,
    credential_provider: Arc<dyn CredentialProvider>,
    config: CloudConfig,
    cloud_state: CloudState,
    openlatch_dir: PathBuf,
    outbox: Option<Arc<Outbox>>,
    shutdown: Option<tokio::sync::watch::Receiver<bool>>,
) {
    let mut rx = rx;
    run_cloud_worker_on(
        &mut rx,
        credential_provider,
        config,
        cloud_state,
        openlatch_dir,
        outbox,
        shutdown,
    )
    .await
}

/// [`run_cloud_worker`] over a **borrowed** receiver.
///
/// The daemon runs the worker under its in-process task supervisor
/// (`core::supervision::task`), which re-invokes the future on every restart.
/// An owned receiver would die with the first panicking run and take every
/// queued event with it; borrowing lets the supervisor park the receiver in an
/// `Arc<tokio::sync::Mutex<_>>` and hand the same channel — queue intact — to
/// the replacement run. Every other caller (tests, non-supervised paths) keeps
/// using [`run_cloud_worker`], which is a thin by-value wrapper around this.
#[allow(clippy::too_many_arguments)]
pub async fn run_cloud_worker_on(
    rx: &mut mpsc::Receiver<CloudEvent>,
    credential_provider: Arc<dyn CredentialProvider>,
    config: CloudConfig,
    cloud_state: CloudState,
    openlatch_dir: PathBuf,
    outbox: Option<Arc<Outbox>>,
    shutdown: Option<tokio::sync::watch::Receiver<bool>>,
) {
    // When the caller supplied no signal, hold a live sender for the worker's
    // whole lifetime: a dropped sender makes `changed()` resolve immediately
    // and forever, which would spin the select! instead of never firing.
    let (_shutdown_keepalive, mut shutdown_rx) = match shutdown {
        Some(rx) => (None, rx),
        None => {
            let (tx, rx) = tokio::sync::watch::channel(false);
            (Some(tx), rx)
        }
    };

    let client = build_cloud_client(&config);
    let credential_poll_interval = Duration::from_millis(config.credential_poll_interval_ms);

    // Durable-outbox drain task. Triggered on daemon startup (first tick
    // below) and on each successful health probe via the shared
    // `cloud_state.drain_notify`. The task exits when the state's Arc is
    // dropped (i.e. when the daemon shuts down).
    if let Some(outbox) = outbox.as_ref() {
        let outbox = outbox.clone();
        let client = client.clone();
        let config = config.clone();
        let provider = credential_provider.clone();
        let state = cloud_state.clone();
        let dir = openlatch_dir.clone();
        let notify = state.drain_notify.clone();
        // Cached once at worker startup: the agent_id is stable for the
        // lifetime of the process (`openlatch init` is a prerequisite to
        // starting the daemon, and the id never rotates without a restart).
        let agent_id = crate::config::sniff_agent_id(&dir).unwrap_or_default();
        tokio::spawn(async move {
            // Volatile per-id attempt counts. Restart wipes the map and the
            // quarantine budget resets per daemon lifetime. Wrapped in
            // `Mutex` so the drain closure can update it across `.await`.
            let attempts: Arc<std::sync::Mutex<HashMap<String, u32>>> =
                Arc::new(std::sync::Mutex::new(HashMap::new()));
            let ctx = OutboxDrainCtx {
                outbox: &outbox,
                client: &client,
                config: &config,
                credential_provider: &provider,
                cloud_state: &state,
                openlatch_dir: &dir,
                agent_id: &agent_id,
                attempts: &attempts,
            };
            // Kick off an immediate drain attempt at startup in case the
            // previous daemon run left entries behind.
            drain_outbox_once(&ctx).await;
            loop {
                notify.notified().await;
                // Pause drains while the live channel is under emergency
                // congestion — replay events would just compete with live
                // hooks for the same headroom. The next health-recovery
                // notify will retry.
                if state.is_emergency_mode() {
                    tracing::debug!(
                        "outbox drain: skipping pass — cloud channel in emergency mode"
                    );
                    continue;
                }
                drain_outbox_once(&ctx).await;
            }
        });
    }

    // Load initial credential via spawn_blocking (keyring calls must not run in async context)
    let provider = credential_provider.clone();
    let mut current_key: Option<SecretString> =
        tokio::task::spawn_blocking(move || provider.retrieve())
            .await
            .unwrap_or(None);
    cloud_state.set_no_credential(current_key.is_none());

    let mut auth_error = false;

    // Publish the starting value immediately.
    //
    // `cloud_state.json` is otherwise only written on a *transition* — the
    // 401/403 latch setting it, or `clear_auth_error_latch` clearing it. A
    // worker always starts at `auth_error = false`, so it never transitions on
    // the way up and never rewrites the file. A daemon that died (or was
    // killed) while latched therefore leaves `{"auth_error":true}` on disk, and
    // every later daemon inherits that claim about itself for as long as it
    // stays healthy — the file outlives the process whose state it describes.
    //
    // Writing it here makes the file mean "the current worker's state" from the
    // moment it boots, rather than "the last transition any worker ever made".
    // Non-fatal: this is a reporting surface, and failing to publish it must
    // never stop the worker from forwarding events.
    if let Err(e) = persist_cloud_state(&openlatch_dir, auth_error) {
        tracing::warn!(
            error = %e,
            "cloud worker: failed to persist initial cloud_state.json"
        );
    }

    // Whether we have already emitted the OL-1200 "no API key" warning for the
    // current missing-credential streak. Reset whenever a credential is loaded,
    // so a disable/re-enable cycle warns again — but a long-lived missing-key
    // state produces one actionable WARN, not one per event.
    let mut missing_key_warned = current_key.is_none();
    // Active degraded-streak latch. Some(reason) → outage in progress; the
    // first failure emitted a WARN, subsequent failures drop to DEBUG. Cleared
    // by `note_recovery` on the next successful POST or health probe, which
    // emits a single INFO with the spooled count.
    let mut degraded: Option<&'static str> = None;
    if current_key.is_none() {
        tracing::warn!(
            code = "OL-1200",
            "cloud worker: no API key available — cloud forwarding disabled until you run \
             'openlatch auth login' or set [cloud] enabled = false in config.toml (fail-open, \
             events are still logged locally)"
        );
    }
    let mut last_credential_poll = tokio::time::Instant::now();
    let mut last_health_tick = tokio::time::Instant::now();
    let mut emergency_recovery_ticks: u32 = 0;

    // ------------------------------------------------------------------
    // Batch accumulator (D43)
    //
    // The deadline is anchored to the FIRST buffered event and is NEVER reset
    // on subsequent pushes.
    //
    // REJECTED — resetting the timer on every push. It is simpler to write,
    // but under any sustained event rate the timer is perpetually reset and
    // the batch only ever flushes on the SIZE trigger, so `batch_max_wait_ms`
    // silently stops bounding latency. That is precisely the bug the PRD's
    // acceptance criterion is written to catch: "the batch SHALL flush no
    // later than batch_max_wait_ms after the FIRST event was enqueued."
    // ------------------------------------------------------------------
    let size_trigger = config.batch_max_events.max(1);
    let mut buf: Vec<CloudEvent> = Vec::with_capacity(size_trigger);
    let mut deadline: Option<tokio::time::Instant> = None;
    let flush_ctx = FlushCtx {
        client: &client,
        config: &config,
        openlatch_dir: &openlatch_dir,
        outbox: outbox.as_ref(),
        cloud_state: &cloud_state,
    };

    loop {
        // Credential hot-reload: check every 60 seconds (D-07)
        if last_credential_poll.elapsed() >= credential_poll_interval {
            let provider = credential_provider.clone();
            let new_key = tokio::task::spawn_blocking(move || provider.retrieve())
                .await
                .unwrap_or(None);

            // Compare by exposing secret (both sides must match)
            let key_changed = match (&current_key, &new_key) {
                (None, None) => false,
                (Some(_), None) | (None, Some(_)) => true,
                (Some(old), Some(new)) => {
                    use secrecy::ExposeSecret;
                    old.expose_secret() != new.expose_secret()
                }
            };

            // The old unconditional reset on every poll caused a 60s flap
            // against permanently-revoked keys: clear → 401 → spool → repeat.
            // Now we only clear on a real key rotation, OR mirror an
            // out-of-band clear from `POST /admin/auth/refresh` so WR-01
            // (same-key re-login) still resumes POSTs immediately.
            if new_key.is_some() {
                let out_of_band_clear = !key_changed && auth_error && !cloud_state.is_auth_error();
                if key_changed {
                    tracing::info!(
                        "cloud worker: credential refreshed — resetting auth_error state"
                    );
                    clear_auth_error_latch(
                        &mut auth_error,
                        &cloud_state,
                        &openlatch_dir,
                        "credential refresh",
                    );
                } else if out_of_band_clear {
                    clear_auth_error_latch(
                        &mut auth_error,
                        &cloud_state,
                        &openlatch_dir,
                        "out-of-band auth clear",
                    );
                }
                missing_key_warned = false;
                current_key = new_key;
            }

            cloud_state.set_no_credential(current_key.is_none());
            last_credential_poll = tokio::time::Instant::now();
        }

        // The timer branch needs a valid instant even when disarmed; the
        // `if deadline.is_some()` guard is what actually disables it.
        let flush_at = deadline.unwrap_or_else(far_future);

        // Receive next event (with timeout to allow periodic credential polling)
        //
        // D44 — CANCEL SAFETY. `tokio::select!` drops the losing branch's
        // future. The concern is that `rx.recv()` might dequeue an item and
        // then be cancelled, losing it. It cannot: `mpsc::Receiver::recv` is
        // documented cancel-safe — "if recv is used as the event in a
        // tokio::select! statement and some other branch completes first, it
        // is guaranteed that no messages were received on this channel."
        // `Sleep` is trivially cancel-safe, and so is `watch::Receiver::changed`.
        //
        // REJECTED — wrapping the recv in a `tokio::spawn` + oneshot to "make
        // it safe". That adds a task per event, introduces a real loss window
        // at shutdown, and solves a problem that does not exist. This note
        // exists so nobody "fixes" it later.
        tokio::select! {
            // `biased;` with the TIMER FIRST. Without it select! polls
            // branches in a random order, so under a continuously-ready
            // channel an already-due timer only wins with some probability
            // per iteration — the flush bound degrades from "guaranteed" to
            // "0.5^k chance of being k iterations late", which is exactly the
            // kind of soft latency erosion D43 exists to rule out. A due timer
            // must always win, deterministically.
            biased;

            _ = tokio::time::sleep_until(flush_at), if deadline.is_some() => {
                flush_batch(
                    &flush_ctx,
                    &mut buf,
                    current_key.as_ref(),
                    "time",
                    &mut auth_error,
                    &mut degraded,
                ).await;
                deadline = None;
                continue;
            }
            // Explicit shutdown. Fires on an actual signal or on the sender
            // being dropped; either way the post-loop flush runs.
            _ = shutdown_rx.changed() => {
                tracing::info!("cloud worker: shutdown signalled, flushing in-flight batch");
                break;
            }
            // Wake up periodically to check credential reload deadline
            _ = tokio::time::sleep_until(
                last_credential_poll + credential_poll_interval
            ) => {
                // Re-enter loop to perform credential poll
                continue;
            }
            // Periodic cloud health probe — clears network-degraded status
            // once the cloud is reachable again, even on quiet machines.
            //
            // Adaptive cadence: 60s while healthy, backing off aggressively
            // (5s → 10s → 20s → 30s cap, +/-20% jitter) while degraded so
            // reconnect after an offline boot is detected within ~5–15s
            // instead of up to 60s.
            _ = tokio::time::sleep_until(
                last_health_tick + health_delay(&cloud_state)
            ) => {
                last_health_tick = tokio::time::Instant::now();
                match cloud_health_check(&client, &config).await {
                    Ok(()) => {
                        note_recovery(&mut degraded, &cloud_state);
                        cloud_state.record_health_ok();
                        // Cloud just recovered — wake the outbox drain task
                        // AND the daemon-side fallback-replay task so any
                        // events captured while offline flow out before the
                        // next live hook arrives.
                        cloud_state.notify_drain();
                    }
                    Err(e) => {
                        cloud_state.record_probe_failure();
                        tracing::debug!(
                            error = %e,
                            consecutive_failures = cloud_state.consecutive_probe_failures(),
                            "cloud health check failed"
                        );
                    }
                }
                // Detector tick shares the health probe cadence — no new task.
                update_emergency_mode(&cloud_state, &config, &mut emergency_recovery_ticks);
                continue;
            }
            maybe = rx.recv() => {
                let Some(event) = maybe else {
                    // Channel closed — all senders dropped, daemon is shutting down
                    tracing::info!("cloud worker: channel closed, exiting");
                    break;
                };

                // The two pre-POST guards run at RECV time, not flush time.
                // While auth is latched or no key is loaded the event is
                // spooled immediately and never enters the buffer, so the
                // buffer only ever holds sendable events and each event keeps
                // the exact spool reason it would have had before batching.
                // Checking at flush time instead would spool whole buffers
                // under a single, wrong reason.
                if auth_error {
                    tracing::debug!(
                        code = "OL-1201",
                        "cloud worker: auth_error active — spooling event to outbox until credential refresh"
                    );
                    spool_event(outbox.as_ref(), &event, SpoolReason::AuthError);
                    continue;
                }
                if current_key.is_none() {
                    if !missing_key_warned {
                        tracing::warn!(
                            code = "OL-1200",
                            "cloud worker: no API key available — cloud forwarding disabled until \
                             you run 'openlatch auth login' or set [cloud] enabled = false in \
                             config.toml (fail-open, events are still logged locally)"
                        );
                        missing_key_warned = true;
                    } else {
                        tracing::debug!(
                            code = "OL-1200",
                            "cloud worker: skipping event — still no API key available"
                        );
                    }
                    spool_event(outbox.as_ref(), &event, SpoolReason::NoCredential);
                    continue;
                }

                if buf.is_empty() {
                    deadline = Some(
                        tokio::time::Instant::now()
                            + Duration::from_millis(config.batch_max_wait_ms),
                    );
                }
                buf.push(event);
                if buf.len() >= size_trigger {
                    flush_batch(
                        &flush_ctx,
                        &mut buf,
                        current_key.as_ref(),
                        "size",
                        &mut auth_error,
                        &mut degraded,
                    ).await;
                    deadline = None;
                }
            }
        }
    }

    // Shutdown flush: anything still buffered is POSTed, and anything that
    // fails to POST lands in the outbox. Nothing is dropped on the floor.
    //
    // The flush is BOUNDED, and the buffer is snapshotted first, because the
    // daemon gives this task only a few seconds before it drops the JoinHandle
    // and the runtime aborts it (`daemon/mod.rs`). `flush_batch` starts with a
    // `mem::take`, and the retry ladder inside it can sleep for a server-
    // supplied `Retry-After` — so an abort part-way through would take the
    // already-taken events with it, leaving them neither delivered nor
    // spooled. That is precisely the loss acceptance C5 forbids.
    //
    // On timeout we spool the snapshot. A batch that actually landed just as
    // the deadline passed is therefore re-sent from the outbox on next start;
    // the platform dedups on `client_event_id`, so the cost of that race is a
    // duplicate, and the cost of the alternative is a lost event. Prefer
    // at-least-once.
    let snapshot = buf.clone();
    if tokio::time::timeout(
        SHUTDOWN_FLUSH_BUDGET,
        flush_batch(
            &flush_ctx,
            &mut buf,
            current_key.as_ref(),
            "shutdown",
            &mut auth_error,
            &mut degraded,
        ),
    )
    .await
    .is_err()
    {
        tracing::warn!(
            code = "OL-1200",
            buffered = snapshot.len(),
            "cloud worker: shutdown flush exceeded its budget — spooling the batch to the outbox"
        );
        for event in &snapshot {
            spool_event(flush_ctx.outbox, event, SpoolReason::Network);
        }
    }
}

/// The parts of the worker's state that a flush needs and never mutates.
/// Bundled so `flush_batch` stays under the argument-count limit and so the
/// borrow set is obvious at the call site.
struct FlushCtx<'a> {
    client: &'a reqwest::Client,
    config: &'a CloudConfig,
    openlatch_dir: &'a Path,
    outbox: Option<&'a Arc<Outbox>>,
    cloud_state: &'a CloudState,
}

/// Post the accumulated buffer as one or more batches, then clear it.
///
/// Splits under both wire caps (D45), preserves the per-error-class behaviour
/// of the pre-batching single-event path verbatim, and — on terminal failure —
/// spools every envelope of the batch individually (D23).
///
/// `reason` is one of `size` / `time` / `shutdown` / `drain` and is logged
/// alongside the batch size at `debug`, which is exactly what the PRD's
/// Telemetry table asks for to tune the two knobs.
async fn flush_batch(
    ctx: &FlushCtx<'_>,
    buf: &mut Vec<CloudEvent>,
    key: Option<&SecretString>,
    reason: &'static str,
    auth_error: &mut bool,
    degraded: &mut Option<&'static str>,
) {
    if buf.is_empty() {
        return;
    }
    let events = std::mem::take(buf);
    tracing::debug!(
        batch_size = events.len(),
        reason,
        "cloud worker: flushing event batch"
    );

    let prepared = prepare_batch(&events, ctx.cloud_state);
    if prepared.is_empty() {
        return;
    }

    let Some(key) = key else {
        // Defensive: the recv-time guard keeps un-sendable events out of the
        // buffer, so this only fires if the credential vanished mid-flight.
        spool_group(ctx.outbox, &prepared, SpoolReason::NoCredential);
        return;
    };

    for group in split_batches(&prepared, ctx.config.batch_max_events) {
        if *auth_error {
            // An earlier group in this same flush latched 401/403. Sending the
            // rest would just burn POSTs against a credential the server has
            // already rejected.
            spool_group(ctx.outbox, group, SpoolReason::AuthError);
            continue;
        }
        let n = group.len() as u64;
        match post_batch(ctx.client, ctx.config, key, group, ctx.openlatch_dir).await {
            Ok(()) => {
                note_recovery(degraded, ctx.cloud_state);
                tracing::debug!(batch_size = n, "cloud worker: batch forwarded successfully");
                // Record the BATCH LENGTH, not one: every counter downstream
                // (`cloud_forwarded_count` on /metrics, `openlatch status`) is
                // per-event and would under-count by up to 100x otherwise.
                ctx.cloud_state.record_successful_forwards(n);
                // A live POST succeeding also signals connectivity — wake
                // the drain listeners so any queued outbox/fallback entries
                // flow out without waiting for the next health tick.
                ctx.cloud_state.notify_drain();
            }
            Err(CloudError::AuthError) => {
                tracing::warn!(
                    code = "OL-1201",
                    batch_size = n,
                    "cloud worker: auth error (401/403) — pausing POSTs until credential refresh"
                );
                *auth_error = true;
                ctx.cloud_state
                    .auth_error
                    .store(true, std::sync::atomic::Ordering::Relaxed);
                if let Err(e) = persist_cloud_state(ctx.openlatch_dir, true) {
                    tracing::warn!(error = %e, "cloud worker: failed to persist cloud_state.json");
                }
                spool_group(ctx.outbox, group, SpoolReason::AuthError);
            }
            Err(CloudError::RateLimit { retry_after_secs }) => {
                // A 429 is routine backpressure for a batched-ingest client, not
                // an outage: back off for Retry-After and retry once. Logged at
                // `debug` and it does NOT latch `degraded` — only actual data
                // deferral escalates (the spool below, then the outbox drain's
                // OL-1205/1207). This keeps a rate-limited boot drain quiet
                // instead of emitting one WARN per throttled batch.
                tracing::debug!(
                    code = "OL-1202",
                    retry_after_secs,
                    batch_size = n,
                    "cloud worker: rate limited (429) — backing off and retrying"
                );
                tokio::time::sleep(Duration::from_secs(retry_after_secs)).await;
                // Retry once after the wait; spool on second failure so the
                // drain task replays it once the rate-limit window clears.
                match post_batch(ctx.client, ctx.config, key, group, ctx.openlatch_dir).await {
                    // A batch that succeeds on the RETRY is still forwarded.
                    // Recording only the first-attempt success makes
                    // `cloud_forwarded_count` (and the `openlatch status` that
                    // reads it) silently undercount every batch that recovered
                    // from a transient failure — which is most of them on a
                    // flaky link.
                    Ok(()) => ctx.cloud_state.record_successful_forwards(n),
                    Err(_) => {
                        ctx.cloud_state.record_drops(n);
                        spool_group(ctx.outbox, group, SpoolReason::RateLimit);
                    }
                }
            }
            Err(CloudError::ServerError) => {
                if degraded.is_none() {
                    tracing::warn!(
                        code = "OL-1200",
                        batch_size = n,
                        "cloud worker: server error (5xx) — retrying once; suppressing further warnings until recovery"
                    );
                    *degraded = Some("server_error");
                } else {
                    tracing::debug!(code = "OL-1200", "cloud worker: 5xx during degraded streak");
                }
                tokio::time::sleep(Duration::from_millis(ctx.config.retry_delay_ms)).await;
                match post_batch(ctx.client, ctx.config, key, group, ctx.openlatch_dir).await {
                    Ok(()) => ctx.cloud_state.record_successful_forwards(n),
                    Err(_) => {
                        ctx.cloud_state.record_drops(n);
                        spool_group(ctx.outbox, group, SpoolReason::ServerError);
                    }
                }
            }
            Err(CloudError::Network) => {
                if degraded.is_none() {
                    tracing::warn!(
                        code = "OL-1200",
                        batch_size = n,
                        "cloud worker: network error — retrying once; suppressing further warnings until recovery"
                    );
                    *degraded = Some("network");
                } else {
                    tracing::debug!(
                        code = "OL-1200",
                        "cloud worker: network error during degraded streak"
                    );
                }
                tokio::time::sleep(Duration::from_millis(ctx.config.retry_delay_ms)).await;
                match post_batch(ctx.client, ctx.config, key, group, ctx.openlatch_dir).await {
                    Ok(()) => ctx.cloud_state.record_successful_forwards(n),
                    Err(_) => {
                        ctx.cloud_state.record_drops(n);
                        spool_group(ctx.outbox, group, SpoolReason::Network);
                    }
                }
            }
            Err(CloudError::ClientError(code)) => {
                // No retry, as before — but log the batch size so a dropped
                // batch is visible as more than a single lost event.
                tracing::warn!(
                    http_status = code,
                    batch_size = n,
                    "cloud worker: unexpected 4xx — dropping batch (no retry)"
                );
                // A dropped batch is N dropped events. Without this the 4xx
                // path is the one failure mode invisible to `cloud_drop_count`
                // — the counter would read zero while events were being
                // discarded.
                ctx.cloud_state.record_drops(n);
            }
        }
    }
}

/// Unauthenticated GET against `{api_url}/api/v1/health`. Used by the worker's
/// background tick to detect cloud recovery without waiting for a hook event.
/// Non-2xx responses and transport errors surface as `Err` so the caller can
/// log at debug level; they never update `drop_count` or flip `auth_error`.
async fn cloud_health_check(
    client: &reqwest::Client,
    config: &CloudConfig,
) -> Result<(), reqwest::Error> {
    let base = config.api_url.trim_end_matches('/');
    let url = format!("{base}/api/v1/health");
    let resp = client.get(&url).send().await?;
    resp.error_for_status().map(|_| ())
}

/// Failure category labels for the `cloud_event_spooled` telemetry event.
/// Enum (rather than `&'static str`) so a typo can't silently ship a
/// bad label, and so the closed set stays reviewable in one place.
#[derive(Debug, Clone, Copy)]
enum SpoolReason {
    Network,
    ServerError,
    RateLimit,
    AuthError,
    NoCredential,
}

impl SpoolReason {
    fn as_str(self) -> &'static str {
        match self {
            SpoolReason::Network => "network",
            SpoolReason::ServerError => "server_error",
            SpoolReason::RateLimit => "rate_limit",
            SpoolReason::AuthError => "auth_error",
            SpoolReason::NoCredential => "no_credential",
        }
    }
}

/// Best-effort append to the durable outbox. Never propagates errors — the
/// outbox is a recovery mechanism, not a correctness mechanism, and
/// instrumentation in the append path already surfaces OL-1204 if the write
/// failed. A None outbox is a silent no-op (outbox feature disabled).
fn spool_event(outbox: Option<&Arc<Outbox>>, event: &CloudEvent, reason: SpoolReason) {
    if outbox.is_none() {
        return;
    }
    spool_envelope(outbox, &stamp_extensions(event), reason);
}

/// Append one already-stamped envelope to the durable outbox.
fn spool_envelope(outbox: Option<&Arc<Outbox>>, envelope: &serde_json::Value, reason: SpoolReason) {
    let Some(outbox) = outbox else { return };
    match outbox.append(envelope) {
        Ok(()) => {
            crate::telemetry::capture_global(crate::telemetry::Event::cloud_event_spooled(
                reason.as_str(),
            ));
        }
        Err(e) => {
            tracing::warn!(
                code = crate::error::ERR_OUTBOX_WRITE_FAILED,
                error = %e,
                path = %outbox.path().display(),
                "cloud worker: failed to spool event to outbox"
            );
        }
    }
}

/// Spool an entire failed batch — **one line per event** (D23).
///
/// REJECTED — spooling the whole batch as one line. Fewer writes, but it
/// changes the outbox file format and breaks three things that operate per
/// entry: the drain path, the per-entry `OUTBOX_MAX_ATTEMPTS` counter, and the
/// quarantine logic (OL-1207). It would also break `openlatch status` and the
/// existing `outbox_max_bytes` drop-oldest eviction.
fn spool_group(outbox: Option<&Arc<Outbox>>, group: &[PreparedEvent], reason: SpoolReason) {
    if outbox.is_none() {
        return;
    }
    for prepared in group {
        spool_envelope(outbox, &prepared.envelope, reason);
    }
}

/// Clone the envelope and stamp the daemon-owned CloudEvents extension
/// attributes onto it. Shared by the spool path and the POST path so both write
/// the same bytes for the same event.
///
/// `clientversion` is stamped **unconditionally**, on the same rule
/// `daemon/handlers.rs::process_envelope` applies to hook envelopes: the
/// attribute identifies the *forwarder*, not the emitter, and this worker is the
/// single egress for every producer. Doing it here rather than in each builder
/// is what makes the attribute total — `boundary/emit.rs` and the two
/// `config_monitor` builders hand-roll their envelopes and omitted it, which
/// left `unified_events.clientversion` NULL for every economics and config row
/// and starved the platform's fleet-readiness materialization.
///
/// Note this deliberately stamps only forwarder identity. The `ol*` policy
/// extensions stay in `handlers.rs` — an `olverdict` on an
/// `ai.openlatch.config.*` event would be a verdict nobody produced.
fn stamp_extensions(event: &CloudEvent) -> serde_json::Value {
    let mut envelope = event.envelope.clone();
    if let Some(obj) = envelope.as_object_mut() {
        obj.insert(
            "agentid".to_string(),
            serde_json::Value::String(event.agent_id.clone()),
        );
        // OPENLATCH_VERSION, not CARGO_PKG_VERSION — see build.rs.
        obj.insert(
            "clientversion".to_string(),
            serde_json::Value::String(env!("OPENLATCH_VERSION").to_string()),
        );
    }
    envelope
}

/// An envelope stamped with `agentid` + `clientversion` and serialized
/// **exactly once**.
///
/// The serialized form is used for both the byte-cap accounting in
/// `split_batches` and the request body in `post_batch`; serializing twice
/// would double the CPU cost of every flush for no benefit. The parsed value
/// is retained so a failed batch can be spooled without re-parsing.
struct PreparedEvent {
    /// Serialized `agentid`-stamped envelope — one element of the batch array.
    json: String,
    /// The same envelope, for the outbox spool path.
    envelope: serde_json::Value,
}

/// Stamp + serialize a single event, or `None` if it can never be sent.
///
/// An event whose own serialized form does not fit in a request body cannot
/// be batched *or* sent alone: the platform rejects it with 413 before it even
/// parses the body, so every retry fails identically. Dropping it here with
/// OL-1002 is the only way to avoid wedging the queue behind it.
fn prepare_one(envelope: serde_json::Value) -> Option<PreparedEvent> {
    let json = match serde_json::to_string(&envelope) {
        Ok(json) => json,
        Err(e) => {
            tracing::warn!(
                code = crate::error::ERR_EVENT_TOO_LARGE,
                error = %e,
                "cloud worker: dropping event that cannot be serialized"
            );
            return None;
        }
    };
    // `[` + element + `]` is the smallest body that can carry it.
    if json.len() + 2 > MAX_BATCH_BYTES {
        // Recurring, expected condition: some sources routinely emit payloads
        // over the cap, and it fails identically on every retry — so this logs
        // at `debug`, not `warn`, to avoid flooding the operator. The drop stays
        // visible via `CloudState::record_drop` (telemetry counts it).
        tracing::debug!(
            code = crate::error::ERR_EVENT_TOO_LARGE,
            bytes = json.len(),
            max_bytes = MAX_BATCH_BYTES,
            "cloud worker: dropping oversized event — it cannot fit in any batch"
        );
        return None;
    }
    Some(PreparedEvent { json, envelope })
}

/// Stamp + serialize a whole buffer, dropping (and counting) anything that
/// can never be sent.
fn prepare_batch(events: &[CloudEvent], cloud_state: &CloudState) -> Vec<PreparedEvent> {
    let mut out = Vec::with_capacity(events.len());
    for event in events {
        match prepare_one(stamp_extensions(event)) {
            Some(prepared) => out.push(prepared),
            None => cloud_state.record_drop(),
        }
    }
    out
}

/// Greedy split under BOTH caps (D45).
///
/// `batch_max_events` is the tuning knob; `MAX_BATCH_EVENTS` /
/// `MAX_BATCH_BYTES` are the platform's hard limits and are non-negotiable.
///
/// Accounting: each envelope is already serialized once, so this sums `len()`
/// and adds the JSON array framing (`[` + `]` + one `,` per element beyond the
/// first). A batch closes when adding the next event *would* exceed the byte
/// cap, or when the element count reaches `min(batch_max_events, 100)`.
///
/// Returns slices into `prepared` — `post_batch` reuses the same bytes for the
/// request body.
fn split_batches(prepared: &[PreparedEvent], batch_max_events: usize) -> Vec<&[PreparedEvent]> {
    let cap = batch_max_events.clamp(1, MAX_BATCH_EVENTS);
    let mut out = Vec::new();
    let mut start = 0usize;
    let mut payload = 0usize;
    let mut count = 0usize;
    for (i, event) in prepared.iter().enumerate() {
        if count > 0 {
            // Framing for `count + 1` elements: two brackets + `count` commas.
            let framed = payload + event.json.len() + 2 + count;
            if count >= cap || framed > MAX_BATCH_BYTES {
                out.push(&prepared[start..i]);
                start = i;
                payload = 0;
                count = 0;
            }
        }
        payload += event.json.len();
        count += 1;
    }
    if count > 0 {
        out.push(&prepared[start..]);
    }
    out
}

/// Shared context for the outbox drain task. All fields are short-lived
/// borrows captured per drain pass; ownership lives on the spawned task.
struct OutboxDrainCtx<'a> {
    outbox: &'a Arc<Outbox>,
    client: &'a reqwest::Client,
    config: &'a CloudConfig,
    credential_provider: &'a Arc<dyn CredentialProvider>,
    cloud_state: &'a CloudState,
    openlatch_dir: &'a Path,
    agent_id: &'a str,
    attempts: &'a Arc<std::sync::Mutex<HashMap<String, u32>>>,
}

/// Drain the outbox once: read queued envelopes in contiguous groups and
/// repost them via the same `post_batch` path used by live events, so replay
/// traffic has the same shape as live traffic. Halts on the first failed group
/// so a transient outage doesn't drain the entire queue into a retry storm.
/// Skips silently when no credential is available.
async fn drain_outbox_once(ctx: &OutboxDrainCtx<'_>) {
    // Pull a fresh credential before every drain so a recent `openlatch auth
    // login` is picked up without waiting for the 60s credential poll.
    let provider = ctx.credential_provider.clone();
    let key: Option<SecretString> = tokio::task::spawn_blocking(move || provider.retrieve())
        .await
        .unwrap_or(None);
    let Some(key) = key else {
        // No credential — drain is a no-op until auth is configured. The
        // daemon will signal `drain_notify` again the next time a health
        // probe succeeds after a credential lands.
        return;
    };

    // Mirror the live path's wire caps so a replayed group is the same shape
    // as a live batch. `post_batch` re-splits under the exact byte cap before
    // sending, so these limits only have to be in the right ballpark.
    let limits = DrainLimits {
        max_entries: ctx.config.batch_max_events.clamp(1, MAX_BATCH_EVENTS),
        max_bytes: MAX_BATCH_BYTES,
    };

    let stats = ctx
        .outbox
        .drain(limits, |envelopes| {
            let client = ctx.client.clone();
            let config = ctx.config.clone();
            let key = key.clone();
            let openlatch_dir = ctx.openlatch_dir.to_path_buf();
            let agent_id = ctx.agent_id.to_string();
            let attempts = ctx.attempts.clone();
            async move {
                let group_len = envelopes.len();
                // One outcome per input entry, positionally.
                let mut outcomes = vec![DrainOutcome::Forwarded; group_len];
                let mut ids: Vec<Option<String>> = Vec::with_capacity(group_len);
                let mut prepared: Vec<PreparedEvent> = Vec::with_capacity(group_len);

                for (idx, envelope) in envelopes.into_iter().enumerate() {
                    ids.push(
                        envelope
                            .get("id")
                            .and_then(|v| v.as_str())
                            .filter(|s| !s.is_empty())
                            .map(str::to_string),
                    );
                    // Spooled envelopes already carry `agentid`; re-stamping
                    // is idempotent and keeps replay identical to the live
                    // path for entries written by an older client.
                    let event = CloudEvent {
                        envelope,
                        agent_id: agent_id.clone(),
                    };
                    match prepare_one(stamp_extensions(&event)) {
                        Some(p) => prepared.push(p),
                        // Can never be sent — advance past it rather than
                        // wedging the queue, and count it as quarantined
                        // rather than delivered.
                        None => outcomes[idx] = DrainOutcome::Quarantined,
                    }
                }

                let mut failure: Option<CloudError> = None;
                for batch in split_batches(&prepared, config.batch_max_events) {
                    match post_batch(&client, &config, &key, batch, &openlatch_dir).await {
                        Ok(()) => {}
                        // ClientError is terminal — treat as drained so the
                        // entries are removed. Everything else keeps them for
                        // the next drain cycle.
                        Err(CloudError::ClientError(code)) => {
                            tracing::warn!(
                                http_status = code,
                                batch_size = batch.len(),
                                "outbox drain: dropping entries on unexpected 4xx"
                            );
                        }
                        Err(e) => {
                            failure = Some(e);
                            break;
                        }
                    }
                }

                let Some(e) = failure else {
                    // Healthy drain: the attempt map is empty in the common
                    // case. A `try_lock + is_empty` check would race with
                    // concurrent inserts; an unconditional `is_empty` peek
                    // under the mutex is the cheap path — one atomic and one
                    // length compare, no allocation.
                    let mut guard = attempts.lock().unwrap();
                    if !guard.is_empty() {
                        for id in ids.iter().flatten() {
                            guard.remove(id);
                        }
                    }
                    return Ok(outcomes);
                };

                // Per-id attempt cap: after OUTBOX_MAX_ATTEMPTS transient
                // failures within a single daemon lifetime an entry is
                // quarantined so the queue can drain past it. Without this,
                // one permanently-rejecting envelope at the head of
                // `outbox.jsonl` blocks every entry behind it.
                //
                // A failed group charges EVERY entry in it an attempt — the
                // group is what gets retried, so the whole group is what pays.
                // The group only advances once every entry has exhausted its
                // budget; until then the pass halts and nothing moves.
                let mut all_exhausted = true;
                {
                    let mut guard = attempts.lock().unwrap();
                    for id in &ids {
                        let Some(id) = id else {
                            // No id to key on — cannot track attempts, so this
                            // group can never be quarantined out of the way.
                            all_exhausted = false;
                            continue;
                        };
                        let entry = guard.entry(id.clone()).or_insert(0);
                        *entry = entry.saturating_add(1);
                        if *entry < OUTBOX_MAX_ATTEMPTS {
                            all_exhausted = false;
                        }
                    }
                }

                if all_exhausted {
                    let mut guard = attempts.lock().unwrap();
                    for id in ids.iter().flatten() {
                        tracing::warn!(
                            code = crate::error::ERR_OUTBOX_QUARANTINED,
                            event_id = %id,
                            attempts = OUTBOX_MAX_ATTEMPTS,
                            error = %e,
                            "outbox: quarantining repeatedly-failing entry"
                        );
                        guard.remove(id);
                    }
                    return Ok(vec![DrainOutcome::Quarantined; group_len]);
                }

                tracing::debug!(
                    error = %e,
                    batch_size = group_len,
                    "outbox drain: halted on transient failure — retaining remaining entries"
                );
                Err(())
            }
        })
        .await;

    match stats {
        Ok(stats) => {
            if stats.drained > 0 || stats.failed > 0 || stats.corrupt > 0 || stats.quarantined > 0 {
                tracing::info!(
                    drained = stats.drained,
                    failed = stats.failed,
                    corrupt = stats.corrupt,
                    quarantined = stats.quarantined,
                    remaining = ctx.outbox.pending_count(),
                    "outbox drain completed"
                );
                crate::telemetry::capture_global(crate::telemetry::Event::cloud_outbox_drained(
                    stats.drained,
                    stats.failed,
                    stats.corrupt,
                    stats.quarantined,
                ));
            }
            if stats.drained > 0 {
                ctx.cloud_state
                    .forwarded_count
                    .fetch_add(stats.drained, std::sync::atomic::Ordering::Relaxed);
            }
            if stats.failed > 0 {
                tracing::warn!(
                    code = crate::error::ERR_OUTBOX_DRAIN_PARTIAL,
                    failed = stats.failed,
                    remaining = ctx.outbox.pending_count(),
                    "outbox drain: partial — some entries could not be replayed yet"
                );
            }
        }
        Err(e) => {
            tracing::warn!(
                code = crate::error::ERR_OUTBOX_DRAIN_PARTIAL,
                error = %e,
                "outbox drain: I/O error while replaying"
            );
        }
    }
}

/// POST one batch of cloud events to the ingestion endpoint.
///
/// Takes an already-prepared slice — the caller stamped `agentid` and
/// serialized each envelope once, and `split_batches` guaranteed the slice is
/// within both wire caps. Attaches headers with a UUIDv7 request ID and
/// interprets the HTTP response status.
///
/// The status mapping is unchanged from the pre-batching single-event path;
/// only the *unit* it operates on went from one event to N.
///
/// # Returns
///
/// - `Ok(())` on 2xx success
/// - `Err(CloudError::AuthError)` on 401/403
/// - `Err(CloudError::RateLimit { ... })` on 429 (parses `Retry-After` header)
/// - `Err(CloudError::ServerError)` on 5xx
/// - `Err(CloudError::Network)` on transport errors
/// - `Err(CloudError::ClientError(code))` on other 4xx
async fn post_batch(
    client: &reqwest::Client,
    config: &CloudConfig,
    key: &SecretString,
    batch: &[PreparedEvent],
    _openlatch_dir: &Path,
) -> Result<(), CloudError> {
    // Generate UUIDv7 request ID (D-19)
    let request_id = uuid::Uuid::now_v7().to_string();

    // CloudEvents v1.0.2 structured-mode batch format: a bare JSON array of
    // envelopes. Assembled from the already-serialized elements rather than
    // re-serializing — `build_cloud_headers` sets
    // `Content-Type: application/cloudevents-batch+json`, which `.body()`
    // leaves alone.
    let mut body = String::with_capacity(batch.iter().map(|e| e.json.len() + 1).sum::<usize>() + 2);
    body.push('[');
    for (i, event) in batch.iter().enumerate() {
        if i > 0 {
            body.push(',');
        }
        body.push_str(&event.json);
    }
    body.push(']');

    // Build headers (D-17)
    let headers = build_cloud_headers(key, &request_id);

    // WR-02: Strip trailing slash to prevent double-slash URLs and guard against
    // query-injection if api_url ends with a query separator.
    let base = config.api_url.trim_end_matches('/');
    let url = format!("{base}/api/v1/events/ingest");

    let response = client
        .post(&url)
        .headers(headers)
        .body(body)
        .send()
        .await
        .map_err(|_| CloudError::Network)?;

    let status = response.status();

    if status.is_success() {
        return Ok(());
    }

    match status.as_u16() {
        401 | 403 => Err(CloudError::AuthError),
        429 => {
            // Parse Retry-After header; default to config value if absent or unparsable
            let retry_after_secs = response
                .headers()
                .get("Retry-After")
                .and_then(|v| v.to_str().ok())
                .and_then(|s| s.parse::<u64>().ok())
                .unwrap_or(config.rate_limit_default_secs);
            Err(CloudError::RateLimit { retry_after_secs })
        }
        500..=599 => Err(CloudError::ServerError),
        code => Err(CloudError::ClientError(code)),
    }
}

/// Write the cloud auth error state to disk for cross-process visibility (CLOUD-08).
///
/// Writes `{openlatch_dir}/cloud_state.json` atomically:
/// 1. Write to `cloud_state.json.tmp`
/// 2. Rename to `cloud_state.json`
///
/// This prevents `openlatch status` (a separate process) from reading a partial file.
/// The parent directory is created if it does not exist.
///
/// # Format
///
/// ```json
/// {"auth_error": true, "updated_at": "2026-04-09T12:00:00Z"}
/// ```
pub fn persist_cloud_state(openlatch_dir: &Path, auth_error: bool) -> std::io::Result<()> {
    // Create parent directory if missing
    std::fs::create_dir_all(openlatch_dir)?;

    let updated_at = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();

    let content = format!(
        "{{\"auth_error\":{},\"updated_at\":\"{}\"}}\n",
        auth_error, updated_at
    );

    let tmp_path = openlatch_dir.join("cloud_state.json.tmp");
    let final_path = openlatch_dir.join("cloud_state.json");

    std::fs::write(&tmp_path, &content)?;
    std::fs::rename(&tmp_path, &final_path)?;

    Ok(())
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use secrecy::SecretString;
    use std::sync::atomic::Ordering;
    use std::sync::Mutex;
    use tokio::sync::mpsc;

    // ---------------------------------------------------------------------------
    // Test credential provider
    // ---------------------------------------------------------------------------

    struct TestCredentialProvider {
        key: Mutex<Option<String>>,
        /// Number of `retrieve()` calls. Lets a test observe that the
        /// credential-poll branch of the `select!` is still running.
        retrievals: std::sync::atomic::AtomicU64,
    }

    impl TestCredentialProvider {
        fn with_key(key: &str) -> Arc<Self> {
            Arc::new(Self {
                key: Mutex::new(Some(key.to_string())),
                retrievals: std::sync::atomic::AtomicU64::new(0),
            })
        }

        fn empty() -> Arc<Self> {
            Arc::new(Self {
                key: Mutex::new(None),
                retrievals: std::sync::atomic::AtomicU64::new(0),
            })
        }

        fn set_key(&self, key: &str) {
            *self.key.lock().unwrap() = Some(key.to_string());
        }

        fn retrievals(&self) -> u64 {
            self.retrievals.load(Ordering::Relaxed)
        }
    }

    impl CredentialProvider for TestCredentialProvider {
        fn retrieve(&self) -> Option<SecretString> {
            self.retrievals.fetch_add(1, Ordering::Relaxed);
            self.key
                .lock()
                .ok()
                .and_then(|g| g.as_ref().map(|k| SecretString::from(k.clone())))
        }
    }

    #[test]
    fn test_update_emergency_mode_engages_after_sustained_streak() {
        use std::time::{SystemTime, UNIX_EPOCH};
        let state = CloudState::new();
        let config = CloudConfig::default();
        let mut recovery_ticks = 0u32;

        // Bump drops above the threshold, then back-date the window-start
        // so the detector sees a >10s streak immediately.
        for _ in 0..(EMERGENCY_DROP_THRESHOLD + 1) {
            state.record_live_drop();
        }
        let now_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;
        state.live_drops_window_start.store(
            now_ms.saturating_sub(EMERGENCY_WINDOW_MS + 1_000),
            Ordering::Relaxed,
        );

        update_emergency_mode(&state, &config, &mut recovery_ticks);
        assert!(
            state.is_emergency_mode(),
            "detector must engage on sustained streak"
        );
    }

    #[test]
    fn test_update_emergency_mode_does_not_engage_on_brief_spike() {
        let state = CloudState::new();
        let config = CloudConfig::default();
        let mut recovery_ticks = 0u32;

        // Drops above threshold but the streak window is fresh (<1ms old) —
        // detector must not engage.
        for _ in 0..(EMERGENCY_DROP_THRESHOLD + 1) {
            state.record_live_drop();
        }
        update_emergency_mode(&state, &config, &mut recovery_ticks);
        assert!(
            !state.is_emergency_mode(),
            "detector must not engage on a brief spike (window too short)"
        );
    }

    #[test]
    fn test_update_emergency_mode_clears_after_two_clean_ticks() {
        let state = CloudState::new();
        let config = CloudConfig::default();
        let mut recovery_ticks = 0u32;

        // Force emergency mode on, drops + high-water cleared (simulating
        // successful POSTs draining the channel below the clear threshold).
        state.set_emergency_mode(true);
        assert!(state.is_emergency_mode());

        // First clean tick: counter advances but flag stays.
        update_emergency_mode(&state, &config, &mut recovery_ticks);
        assert!(state.is_emergency_mode());
        assert_eq!(recovery_ticks, 1);

        // Second clean tick: flag clears.
        update_emergency_mode(&state, &config, &mut recovery_ticks);
        assert!(!state.is_emergency_mode());
    }

    #[test]
    fn test_update_emergency_mode_engages_on_sustained_high_water() {
        let state = CloudState::new();
        let config = CloudConfig::default();
        let mut recovery_ticks = 0u32;

        // Back-date the high-water window so the detector sees a >10s streak
        // *without* a live-drop streak — proves the new trigger path works
        // independently of consecutive_live_drops.
        let now_ms = super::super::now_unix_ms();
        state.channel_high_water_start_ms.store(
            now_ms.saturating_sub(HIGH_WATER_WINDOW_MS + 1_000),
            Ordering::Relaxed,
        );
        assert_eq!(state.consecutive_live_drops(), 0);

        update_emergency_mode(&state, &config, &mut recovery_ticks);
        assert!(
            state.is_emergency_mode(),
            "high-water sustained should engage emergency mode without live drops"
        );
    }

    #[test]
    fn test_update_emergency_mode_does_not_clear_while_high_water_active() {
        let state = CloudState::new();
        let config = CloudConfig::default();
        let mut recovery_ticks = 0u32;

        // Emergency mode on, drops cleared, but high-water still latched
        // (start stamped a few ms in the past so `window_ms` is non-zero).
        // Recovery debounce must NOT advance.
        state.set_emergency_mode(true);
        let now_ms = super::super::now_unix_ms();
        state
            .channel_high_water_start_ms
            .store(now_ms.saturating_sub(500), Ordering::Relaxed);

        update_emergency_mode(&state, &config, &mut recovery_ticks);
        assert!(state.is_emergency_mode());
        assert_eq!(
            recovery_ticks, 0,
            "recovery debounce must stay at 0 while high-water remains"
        );
    }

    #[test]
    fn test_build_cloud_client_creates_client_with_pool_max_idle_per_host() {
        // build_cloud_client must succeed and produce a working client
        let config = CloudConfig::default();
        let _client = build_cloud_client(&config);
        // The client is valid if no panic occurred — reqwest doesn't expose internals for assertion
    }

    #[tokio::test]
    async fn test_worker_exits_cleanly_when_channel_closed() {
        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-key");
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();

        // Drop the sender immediately — worker should detect closed channel and exit
        drop(tx);

        // Run worker; it should exit (not hang)
        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            CloudConfig::default(),
            state,
            dir.path().to_path_buf(),
            None,
            None,
        ));

        // Give the worker 1 second to exit
        let result = tokio::time::timeout(std::time::Duration::from_secs(1), handle).await;
        assert!(
            result.is_ok(),
            "worker must exit when channel is closed (timed out waiting)"
        );
    }

    #[tokio::test]
    async fn test_worker_skips_posts_when_no_credential_available() {
        // Worker with no credential — no HTTP calls should be made
        // (We verify indirectly: no panic, events are skipped, worker keeps running)
        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::empty();
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            CloudConfig::default(),
            state,
            dir.path().to_path_buf(),
            None,
            None,
        ));

        // Send an event — worker will skip it (no key) without panicking
        let _ = tx
            .send(CloudEvent {
                envelope: serde_json::json!({}),
                agent_id: "agt_test".to_string(),
            })
            .await;

        // Close channel
        drop(tx);

        let result = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
        assert!(result.is_ok(), "worker must exit cleanly");
    }

    #[test]
    fn test_persist_cloud_state_writes_valid_json_with_auth_error_true() {
        let dir = tempfile::tempdir().unwrap();
        persist_cloud_state(dir.path(), true).expect("persist must succeed");

        let content = std::fs::read_to_string(dir.path().join("cloud_state.json"))
            .expect("cloud_state.json must exist");

        let parsed: serde_json::Value = serde_json::from_str(&content).expect("must be valid JSON");

        assert_eq!(
            parsed["auth_error"], true,
            "auth_error must be true: {content}"
        );
        assert!(
            parsed["updated_at"].as_str().is_some(),
            "updated_at must be present: {content}"
        );
    }

    #[test]
    fn test_persist_cloud_state_writes_valid_json_with_auth_error_false() {
        let dir = tempfile::tempdir().unwrap();
        persist_cloud_state(dir.path(), false).expect("persist must succeed");

        let content = std::fs::read_to_string(dir.path().join("cloud_state.json"))
            .expect("cloud_state.json must exist");

        let parsed: serde_json::Value = serde_json::from_str(&content).expect("must be valid JSON");
        assert_eq!(parsed["auth_error"], false);
    }

    #[test]
    fn test_persist_cloud_state_creates_parent_directory_if_missing() {
        let base = tempfile::tempdir().unwrap();
        let nested = base.path().join("a").join("b").join("c");
        // Directory does not exist yet
        assert!(!nested.exists());

        persist_cloud_state(&nested, false).expect("must create directories and write");

        assert!(nested.join("cloud_state.json").exists());
    }

    #[tokio::test]
    async fn test_worker_auth_error_set_when_credential_available_but_server_returns_401() {
        // This test verifies the auth_error flag is set after a 401 response.
        // We use a mock server that returns 401.
        use std::sync::atomic::Ordering;

        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(401)
            .with_body("{}")
            .create_async()
            .await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();

        let config = CloudConfig {
            api_url: server.url(),
            // Batching is on by default (50 events / 5s). This test sends a
            // single event and asserts on the POST before closing the
            // channel, so it needs the time trigger to fire promptly.
            batch_max_wait_ms: 50,
            ..Default::default()
        };

        let state_clone = state.clone();
        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state_clone,
            dir.path().to_path_buf(),
            None,
            None,
        ));

        // Send an event — should trigger 401, set auth_error
        let _ = tx
            .send(CloudEvent {
                envelope: serde_json::json!({"id": "evt_test"}),
                agent_id: "agt_test".to_string(),
            })
            .await;

        // Give worker time to process the event
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        // auth_error flag must be set
        assert!(
            state.auth_error.load(Ordering::Relaxed),
            "auth_error must be true after 401 response"
        );

        // cloud_state.json must exist and contain auth_error: true
        let state_path = dir.path().join("cloud_state.json");
        assert!(state_path.exists(), "cloud_state.json must be written");
        let content = std::fs::read_to_string(&state_path).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&content).unwrap();
        assert_eq!(parsed["auth_error"], true);

        // Cleanup
        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), handle).await;
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn worker_startup_clears_a_stale_auth_error_left_by_a_dead_daemon() {
        // A daemon that is killed while latched leaves {"auth_error":true} on
        // disk. The replacement worker starts at auth_error=false and never
        // transitions, so without an explicit write at startup it would inherit
        // — and keep publishing — the dead daemon's claim about itself.
        let dir = tempfile::tempdir().unwrap();
        let state_path = dir.path().join("cloud_state.json");
        persist_cloud_state(dir.path(), true).unwrap();
        let stale: serde_json::Value =
            serde_json::from_str(&std::fs::read_to_string(&state_path).unwrap()).unwrap();
        assert_eq!(
            stale["auth_error"], true,
            "precondition: stale latch on disk"
        );

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let state = CloudState::new();

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            CloudConfig::default(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        // No events sent: the correction must come from starting up, not from
        // a successful POST.
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        let parsed: serde_json::Value =
            serde_json::from_str(&std::fs::read_to_string(&state_path).unwrap()).unwrap();
        assert_eq!(
            parsed["auth_error"], false,
            "a freshly started worker must publish its own state, not inherit \
             the previous daemon's latch"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), handle).await;
    }

    #[tokio::test]
    async fn test_worker_retries_once_on_5xx_then_drops() {
        // Mock server that returns 500 — worker should retry exactly once (2 requests total)
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(500)
            .with_body("{}")
            .expect(2) // exactly 2 requests: original + 1 retry
            .create_async()
            .await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();

        let config = CloudConfig {
            api_url: server.url(),
            retry_delay_ms: 10, // Fast retry for tests
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state,
            dir.path().to_path_buf(),
            None,
            None,
        ));

        let _ = tx
            .send(CloudEvent {
                envelope: serde_json::json!({}),
                agent_id: "agt_test".to_string(),
            })
            .await;

        // Wait for the retry cycle (10ms delay + processing time)
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), handle).await;
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn test_worker_honors_retry_after_header_on_429() {
        // Mock returns 429 with Retry-After: 1 — worker should wait then retry
        let mut server = mockito::Server::new_async().await;
        // First call: 429 with Retry-After
        let mock_429 = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(429)
            .with_header("Retry-After", "1")
            .with_body("{}")
            .expect(1)
            .create_async()
            .await;
        // Retry call: 200 success
        let mock_200 = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(200)
            .with_body("{\"status\":\"accepted\"}")
            .expect(1)
            .create_async()
            .await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();

        let config = CloudConfig {
            api_url: server.url(),
            // Batching is on by default (50 events / 5s). This test sends a
            // single event and asserts on the POST before closing the
            // channel, so it needs the time trigger to fire promptly.
            batch_max_wait_ms: 50,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state,
            dir.path().to_path_buf(),
            None,
            None,
        ));

        let _ = tx
            .send(CloudEvent {
                envelope: serde_json::json!({}),
                agent_id: "agt_test".to_string(),
            })
            .await;

        // Wait for retry cycle: 1s Retry-After + processing overhead
        tokio::time::sleep(std::time::Duration::from_millis(1500)).await;

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), handle).await;
        mock_429.assert_async().await;
        mock_200.assert_async().await;
    }

    #[tokio::test]
    async fn test_health_tick_clears_consecutive_drops_on_2xx() {
        let mut server = mockito::Server::new_async().await;
        let mock_health = server
            .mock("GET", "/api/v1/health")
            .with_status(200)
            .with_body(r#"{"status":"ok"}"#)
            .expect_at_least(1)
            .create_async()
            .await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let state = CloudState::new();
        state.record_drop();
        state.record_drop();
        state.record_drop();
        assert_eq!(state.consecutive_drops(), 3);
        assert_eq!(state.drop_count(), 3);
        let dir = tempfile::tempdir().unwrap();

        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };

        let state_clone = state.clone();
        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state_clone,
            dir.path().to_path_buf(),
            None,
            None,
        ));

        // Health tick fires every 50ms in test builds — wait long enough for at least one.
        tokio::time::sleep(std::time::Duration::from_millis(250)).await;

        assert_eq!(
            state.consecutive_drops(),
            0,
            "health tick must clear consecutive_drops on 2xx"
        );
        assert_eq!(
            state.drop_count(),
            3,
            "lifetime drop_count must be preserved"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), handle).await;
        mock_health.assert_async().await;
    }

    #[tokio::test]
    async fn test_health_tick_does_not_clear_on_5xx() {
        let mut server = mockito::Server::new_async().await;
        let mock_health = server
            .mock("GET", "/api/v1/health")
            .with_status(500)
            .with_body("{}")
            .expect_at_least(1)
            .create_async()
            .await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let state = CloudState::new();
        state.record_drop();
        state.record_drop();
        let dir = tempfile::tempdir().unwrap();

        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };

        let state_clone = state.clone();
        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state_clone,
            dir.path().to_path_buf(),
            None,
            None,
        ));

        tokio::time::sleep(std::time::Duration::from_millis(250)).await;

        assert_eq!(
            state.consecutive_drops(),
            2,
            "failed health probe must not reset consecutive_drops"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), handle).await;
        mock_health.assert_async().await;
    }

    #[tokio::test]
    async fn test_successful_post_clears_consecutive_drops() {
        let mut server = mockito::Server::new_async().await;
        let _mock_health = server
            .mock("GET", "/api/v1/health")
            .with_status(500)
            .with_body("{}")
            .create_async()
            .await;
        let mock_ingest = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(200)
            .with_body(r#"{"status":"accepted"}"#)
            .expect(1)
            .create_async()
            .await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let state = CloudState::new();
        state.record_drop();
        state.record_drop();
        state.record_drop();
        let dir = tempfile::tempdir().unwrap();

        let config = CloudConfig {
            api_url: server.url(),
            // Batching is on by default (50 events / 5s). This test sends a
            // single event and asserts on the POST before closing the
            // channel, so it needs the time trigger to fire promptly.
            batch_max_wait_ms: 50,
            ..Default::default()
        };

        let state_clone = state.clone();
        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state_clone,
            dir.path().to_path_buf(),
            None,
            None,
        ));

        let _ = tx
            .send(CloudEvent {
                envelope: serde_json::json!({"id": "evt_test"}),
                agent_id: "agt_test".to_string(),
            })
            .await;

        tokio::time::sleep(std::time::Duration::from_millis(200)).await;

        assert_eq!(
            state.consecutive_drops(),
            0,
            "successful forward must clear consecutive_drops"
        );
        assert_eq!(state.drop_count(), 3, "lifetime drop_count preserved");
        assert_eq!(state.forwarded_count(), 1);

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), handle).await;
        mock_ingest.assert_async().await;
    }

    /// Regression: when the worker starts with no credential and receives
    /// events, it must skip them silently after the first warning, then pick
    /// up a late-arriving credential via the hot-reload tick and resume POSTs.
    ///
    /// The observable assertion is that exactly one HTTP POST reaches the
    /// ingestion endpoint — the three events sent before the credential was
    /// loaded must not hit the server.
    #[tokio::test]
    async fn test_worker_recovers_when_credential_appears_after_startup() {
        let mut server = mockito::Server::new_async().await;
        let mock_ingest = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(200)
            .with_body(r#"{"status":"accepted"}"#)
            .expect(1)
            .create_async()
            .await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::empty();
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();

        let config = CloudConfig {
            api_url: server.url(),
            // 50ms reload so the test can flip the credential and observe
            // recovery without a real-time wait.
            credential_poll_interval_ms: 50,
            ..Default::default()
        };

        let provider_handle = provider.clone();
        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        // Three events arrive while the credential is missing — all must be
        // skipped without reaching the mock server.
        for i in 0..3 {
            let _ = tx
                .send(CloudEvent {
                    envelope: serde_json::json!({"id": format!("evt_{i}")}),
                    agent_id: "agt_test".to_string(),
                })
                .await;
        }
        tokio::time::sleep(std::time::Duration::from_millis(30)).await;
        assert_eq!(
            state.forwarded_count(),
            0,
            "events must not be forwarded before a credential is available"
        );

        // Credential appears — the 50ms test-mode reload tick will pick it up.
        provider_handle.set_key("late-arriving-key");
        tokio::time::sleep(std::time::Duration::from_millis(120)).await;

        // Next event must actually be POSTed.
        let _ = tx
            .send(CloudEvent {
                envelope: serde_json::json!({"id": "evt_after_reload"}),
                agent_id: "agt_test".to_string(),
            })
            .await;
        tokio::time::sleep(std::time::Duration::from_millis(150)).await;

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(1), handle).await;
        mock_ingest.assert_async().await;
        assert_eq!(
            state.forwarded_count(),
            1,
            "exactly one event should be forwarded after credential hot-reload"
        );
    }

    /// End-to-end poison-pill: when the same outbox entry fails on every
    /// drain pass, the worker must quarantine it after OUTBOX_MAX_ATTEMPTS
    /// so the queue can drain past it. Without this behaviour, a single
    /// permanently-rejecting envelope at the head of `outbox.jsonl` blocks
    /// every entry behind it forever (the pre-fix bug).
    #[tokio::test]
    async fn test_outbox_quarantines_on_persistent_5xx() {
        use crate::core::cloud::outbox::Outbox;

        let mut server = mockito::Server::new_async().await;
        // Every POST returns 500. We expect: `OUTBOX_MAX_ATTEMPTS` attempts
        // per envelope on the live path (1 initial + 1 retry inside
        // post_event, multiplied across the 5-attempt budget) plus drain
        // retries until quarantine kicks in. We accept any number ≥ the
        // minimum to avoid timing flakiness — the important assertion is
        // that `pending_count()` returns to 0.
        let _mock_ingest = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(500)
            .with_body("boom")
            .expect_at_least(2)
            .create_async()
            .await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));

        let config = CloudConfig {
            api_url: server.url(),
            retry_delay_ms: 5,
            ..Default::default()
        };

        // Pre-seed the outbox with one envelope so we observe the drain
        // task's quarantine path directly (rather than the live retry path).
        outbox
            .append(&serde_json::json!({"id": "evt_poison"}))
            .unwrap();
        assert_eq!(outbox.pending_count(), 1);

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state.clone(),
            dir.path().to_path_buf(),
            Some(outbox.clone()),
            None,
        ));

        // Trigger repeated drain attempts via successive notify_drain. The
        // health tick fires every 50ms in test builds, and each failed POST
        // notifies again, so within ~1s the per-id attempt budget exhausts.
        for _ in 0..(OUTBOX_MAX_ATTEMPTS as usize + 4) {
            state.notify_drain();
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;

        assert_eq!(
            outbox.pending_count(),
            0,
            "quarantine must drain the outbox past the poison entry"
        );
    }

    /// End-to-end outbox replay: when the first POST fails, the envelope
    /// must land in `outbox.jsonl`. When the cloud later recovers, the
    /// drain task fired from a successful health probe must flush it. The
    /// observable assertions are:
    /// - Mockito receives exactly two successful ingest calls: one for the
    ///   health recovery tick (a GET) and one for the replayed event.
    /// - `outbox.jsonl` is empty after recovery.
    /// - `forwarded_count` reflects the replayed event.
    #[tokio::test]
    async fn test_outbox_spools_on_failure_and_drains_on_recovery() {
        use crate::core::cloud::outbox::Outbox;

        // First request returns 500 (live POST fails). Second wave (health
        // probe GET + replayed POST) returns 200.
        let mut server = mockito::Server::new_async().await;
        let mock_fail_ingest = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(500)
            .with_body("boom")
            .expect(2) // live attempt + 1 retry inside the flush
            .create_async()
            .await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));

        let config = CloudConfig {
            api_url: server.url(),
            // Batching is on by default (50 events / 5s). This test sends a
            // single event and asserts on the POST before closing the
            // channel, so it needs the time trigger to fire promptly.
            batch_max_wait_ms: 50,
            retry_delay_ms: 10,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state.clone(),
            dir.path().to_path_buf(),
            Some(outbox.clone()),
            None,
        ));

        // Live POST fails twice (initial + retry) → envelope lands in outbox.
        let _ = tx
            .send(CloudEvent {
                envelope: serde_json::json!({"id": "evt_offline_1"}),
                agent_id: "agt_test".to_string(),
            })
            .await;
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
        mock_fail_ingest.assert_async().await;
        assert_eq!(outbox.pending_count(), 1, "envelope must be spooled");
        assert_eq!(state.forwarded_count(), 0);

        // Cloud recovers. Swap mockito: health probe returns 200, and the
        // drain task's replay POST also returns 200.
        server.reset();
        let mock_health = server
            .mock("GET", "/api/v1/health")
            .with_status(200)
            .expect_at_least(1)
            .create_async()
            .await;
        let mock_replay = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(200)
            .with_body(r#"{"status":"accepted"}"#)
            .expect(1)
            .create_async()
            .await;

        // The test-mode health tick is 50ms. Give the drain task ample
        // time to run after recovery signal fires.
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;

        mock_health.assert_async().await;
        mock_replay.assert_async().await;
        assert_eq!(
            outbox.pending_count(),
            0,
            "outbox must be drained after cloud recovery"
        );
        assert!(
            state.forwarded_count() >= 1,
            "forwarded_count must reflect the replayed event, got {}",
            state.forwarded_count()
        );
    }

    // =================================================================
    // Batching (plan 03)
    //
    // The assertions below are on REQUEST COUNT, not just event delivery.
    // A worker that posts one request per event still delivers every event;
    // only the request count catches the regression.
    // =================================================================

    /// Request bodies captured from the ingest endpoint, in arrival order.
    type Captured = Arc<Mutex<Vec<String>>>;

    /// Mount `POST /api/v1/events/ingest` and record every request body.
    ///
    /// `with_body_from_request` is the only hook mockito offers that runs once
    /// per matched request and can see the request body; the `Vec<u8>` it
    /// returns is the response the worker reads back.
    async fn capture_ingest(
        server: &mut mockito::ServerGuard,
        status: usize,
    ) -> (mockito::Mock, Captured) {
        let captured: Captured = Arc::new(Mutex::new(Vec::new()));
        let sink = captured.clone();
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(status)
            .with_body_from_request(move |req| {
                let body = req
                    .utf8_lossy_body()
                    .map(|b| b.to_string())
                    .unwrap_or_default();
                sink.lock().unwrap().push(body);
                br#"{"status":"accepted"}"#.to_vec()
            })
            .expect_at_least(0)
            .create_async()
            .await;
        (mock, captured)
    }

    /// Mount a permissive, always-failing health probe.
    ///
    /// `health_delay` is `#[cfg(test)]`-overridden to 50ms and
    /// `cloud_health_check` GETs this path on the SAME mockito base URL. A
    /// 2xx probe calls `notify_drain()`, which wakes the outbox drain task and
    /// issues EXTRA POSTs to the ingest endpoint — silently corrupting every
    /// request-count assertion here. Returning 500 keeps the branch exercised
    /// without that side effect.
    async fn quiet_health(server: &mut mockito::ServerGuard) -> mockito::Mock {
        server
            .mock("GET", "/api/v1/health")
            .with_status(500)
            .with_body("{}")
            .expect_at_least(0)
            .create_async()
            .await
    }

    fn evt(id: &str) -> CloudEvent {
        CloudEvent {
            envelope: serde_json::json!({"id": id, "specversion": "1.0"}),
            agent_id: "agt_test".to_string(),
        }
    }

    /// Element counts of each captured batch, in arrival order.
    fn batch_sizes(captured: &Captured) -> Vec<usize> {
        captured
            .lock()
            .unwrap()
            .iter()
            .map(|body| {
                serde_json::from_str::<Vec<serde_json::Value>>(body)
                    .expect("every request body must be a JSON array")
                    .len()
            })
            .collect()
    }

    /// The PRD's named case: 120 events at `batch_max_events = 50` is
    /// **three** requests, not 120. The time trigger is pushed out of reach so
    /// this isolates the size trigger plus the shutdown flush.
    #[tokio::test]
    async fn exact_request_count_for_n_events() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(200);
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 50,
            batch_max_wait_ms: 600_000,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        for i in 0..120 {
            tx.send(evt(&format!("evt_{i:03}"))).await.unwrap();
        }
        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle).await;

        assert_eq!(
            batch_sizes(&captured),
            vec![50, 50, 20],
            "120 events at batch_max_events=50 must be exactly 3 requests"
        );
        assert_eq!(state.forwarded_count(), 120);
    }

    /// Under the size trigger, the time trigger still gets the batch out.
    #[tokio::test]
    async fn flushes_on_time_when_under_size() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 50,
            batch_max_wait_ms: 150,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        for i in 0..3 {
            tx.send(evt(&format!("evt_{i}"))).await.unwrap();
        }
        // Well past the deadline, well short of the 50-event size trigger.
        tokio::time::sleep(std::time::Duration::from_millis(600)).await;

        assert_eq!(
            batch_sizes(&captured),
            vec![3],
            "3 events must leave as one time-triggered request"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    /// **The D43 regression test.** The deadline is anchored to the FIRST
    /// buffered event and is never reset by later ones.
    ///
    /// Timeline with `batch_max_wait_ms = 1000`: push at t=0, push at t=700,
    /// observe at t=1200. Anchored to the first event the flush is due at
    /// t=1000 and has happened. Reset on every push it would be due at
    /// t=1700 and nothing would have been sent — which is exactly how
    /// `batch_max_wait_ms` silently stops bounding latency under load.
    #[tokio::test]
    async fn deadline_anchored_to_first_item() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 50,
            batch_max_wait_ms: 1000,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        tx.send(evt("evt_first")).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(700)).await;
        tx.send(evt("evt_second")).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;

        assert_eq!(
            batch_sizes(&captured),
            vec![2],
            "the batch must flush 1000ms after the FIRST event, not after the last"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    /// No request may exceed either wire cap. The payloads here are sized so
    /// the BYTE cap bites first — capping on element count alone would
    /// produce a body the platform rejects with 413 before it parses it.
    #[tokio::test]
    async fn no_request_exceeds_caps() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(400);
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 100,
            batch_max_wait_ms: 600_000,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        let padding = "x".repeat(4096);
        for i in 0..300 {
            tx.send(CloudEvent {
                envelope: serde_json::json!({"id": format!("evt_{i:03}"), "data": padding}),
                agent_id: "agt_test".to_string(),
            })
            .await
            .unwrap();
        }
        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(10), handle).await;

        let bodies = captured.lock().unwrap().clone();
        assert!(!bodies.is_empty(), "events must have been forwarded");
        let mut total = 0usize;
        let mut byte_capped = false;
        for body in &bodies {
            assert!(
                body.len() <= MAX_BATCH_BYTES,
                "request body of {} bytes exceeds the 256KB cap",
                body.len()
            );
            let elements: Vec<serde_json::Value> = serde_json::from_str(body).unwrap();
            assert!(
                elements.len() <= MAX_BATCH_EVENTS,
                "request carried {} events, over the 100-event cap",
                elements.len()
            );
            if elements.len() < 100 {
                byte_capped = true;
            }
            total += elements.len();
        }
        assert_eq!(total, 300, "every event must still be delivered");
        assert!(
            byte_capped,
            "4KB payloads must close batches on the byte cap before the 100-event cap"
        );
    }

    /// An event too large to fit in any batch is dropped with OL-1002 rather
    /// than wedging the queue behind a POST that fails identically forever.
    /// The events behind it still flush.
    #[tokio::test]
    async fn oversized_single_event_dropped() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 50,
            batch_max_wait_ms: 100,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        tx.send(CloudEvent {
            envelope: serde_json::json!({"id": "evt_huge", "data": "x".repeat(300_000)}),
            agent_id: "agt_test".to_string(),
        })
        .await
        .unwrap();
        tx.send(evt("evt_ok_1")).await.unwrap();
        tx.send(evt("evt_ok_2")).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;

        assert_eq!(
            batch_sizes(&captured),
            vec![2],
            "only the two normal events may be posted — the queue must not wedge"
        );
        assert_eq!(
            state.drop_count(),
            1,
            "the oversized event counts as a drop"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    /// D23 — a failed batch spools EVERY event individually. Asserted by LINE
    /// COUNT, not by file presence: spooling the batch as one line would
    /// satisfy "the file exists" while breaking the drain, the per-entry
    /// attempt counter and the quarantine logic.
    #[tokio::test]
    async fn failed_batch_spools_every_event() {
        use crate::core::cloud::outbox::Outbox;

        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 500).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(20);
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 50,
            batch_max_wait_ms: 100,
            retry_delay_ms: 10,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            state.clone(),
            dir.path().to_path_buf(),
            Some(outbox.clone()),
            None,
        ));

        for i in 0..7 {
            tx.send(evt(&format!("evt_{i}"))).await.unwrap();
        }
        tokio::time::sleep(std::time::Duration::from_millis(600)).await;

        assert_eq!(
            batch_sizes(&captured),
            vec![7, 7],
            "one batch attempt plus exactly one retry"
        );
        let body = std::fs::read_to_string(outbox.path()).expect("outbox.jsonl must exist");
        assert_eq!(
            body.lines().count(),
            7,
            "a failed batch of 7 must spool 7 LINES, not one batch line: {body}"
        );
        for line in body.lines() {
            serde_json::from_str::<serde_json::Value>(line)
                .expect("every spooled line must be a standalone envelope");
        }
        assert_eq!(state.drop_count(), 7, "a spooled batch of 7 is 7 drops");

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    /// The drain re-batches: 120 seeded outbox entries replay as 3 requests,
    /// not 120. Without the group drain in `outbox.rs` this is impossible —
    /// the callback used to receive one entry at a time.
    #[tokio::test]
    async fn outbox_drain_rebatches() {
        use crate::core::cloud::outbox::Outbox;

        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));
        for i in 0..120 {
            outbox
                .append(&serde_json::json!({"id": format!("evt_{i:03}")}))
                .unwrap();
        }

        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 50,
            batch_max_wait_ms: 600_000,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            CloudState::new(),
            dir.path().to_path_buf(),
            Some(outbox.clone()),
            None,
        ));

        // The drain task fires once at worker startup; no live events are
        // sent, so every request below comes from the replay path.
        tokio::time::sleep(std::time::Duration::from_millis(800)).await;

        assert_eq!(
            batch_sizes(&captured),
            vec![50, 50, 20],
            "120 outbox entries must replay as 3 requests"
        );
        assert_eq!(outbox.pending_count(), 0, "the outbox must be empty");

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    /// C5 — nothing is lost at shutdown. The explicit signal (not channel
    /// closure) is what fires here, because `cloud_tx` clones outlive the
    /// daemon's own reference.
    #[tokio::test]
    async fn shutdown_flushes_in_flight() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(20);
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 50,
            batch_max_wait_ms: 600_000,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            state.clone(),
            dir.path().to_path_buf(),
            None,
            Some(shutdown_rx),
        ));

        for i in 0..7 {
            tx.send(evt(&format!("evt_{i}"))).await.unwrap();
        }
        tokio::time::sleep(std::time::Duration::from_millis(150)).await;
        assert!(
            captured.lock().unwrap().is_empty(),
            "the batch must still be in flight — neither trigger has fired"
        );

        // The sender is deliberately still alive: channel closure is NOT what
        // ends the worker here.
        shutdown_tx.send(true).unwrap();
        let exited = tokio::time::timeout(std::time::Duration::from_secs(3), handle).await;
        assert!(exited.is_ok(), "worker must exit on the shutdown signal");

        assert_eq!(
            batch_sizes(&captured),
            vec![7],
            "all 7 buffered events must flush at shutdown — zero lost"
        );
        assert_eq!(state.forwarded_count(), 7);
        drop(tx);
    }

    /// D44 — the `select!` loses nothing across randomised arrival timing.
    /// Delivered + spooled must equal what was enqueued.
    #[tokio::test]
    async fn select_loop_loses_nothing() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(100);
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 50,
            batch_max_wait_ms: 25,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        // Deterministic pseudo-random pacing: bursts, yields and short sleeps
        // so events land on every combination of "timer armed / buffer
        // partially full / flush in flight".
        for i in 0..1000u32 {
            tx.send(evt(&format!("evt_{i:04}"))).await.unwrap();
            match i % 7 {
                0 => tokio::time::sleep(std::time::Duration::from_millis(1)).await,
                3 => tokio::task::yield_now().await,
                _ => {}
            }
        }
        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(30), handle).await;

        let delivered: usize = batch_sizes(&captured).iter().sum();
        assert_eq!(delivered, 1000, "every enqueued event must be delivered");
        assert_eq!(state.forwarded_count(), 1000);
    }

    /// Every element of every batch carries the `agentid` extension —
    /// stamping is per envelope, not per request.
    #[tokio::test]
    async fn agentid_stamped_on_every_envelope() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(20);
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 3,
            batch_max_wait_ms: 600_000,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        for i in 0..9 {
            tx.send(CloudEvent {
                envelope: serde_json::json!({"id": format!("evt_{i}")}),
                agent_id: format!("agt_{i}"),
            })
            .await
            .unwrap();
        }
        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle).await;

        let bodies = captured.lock().unwrap().clone();
        assert_eq!(
            bodies.len(),
            3,
            "9 events at batch_max_events=3 is 3 requests"
        );
        let mut seen = 0;
        for body in &bodies {
            let elements: Vec<serde_json::Value> = serde_json::from_str(body).unwrap();
            for element in elements {
                let id = element["id"].as_str().unwrap();
                let n = id.trim_start_matches("evt_");
                assert_eq!(
                    element["agentid"].as_str(),
                    Some(format!("agt_{n}").as_str()),
                    "every envelope must carry its own agentid: {element}"
                );
                seen += 1;
            }
        }
        assert_eq!(seen, 9);
    }

    /// `clientversion` is stamped at egress even when the producer never set
    /// it. This is the regression test for the economics + config streams:
    /// `boundary/emit.rs` and both `config_monitor` builders hand-roll their
    /// envelopes without the attribute, which left
    /// `unified_events.clientversion` NULL for every row they produced.
    #[test]
    fn clientversion_stamped_when_producer_omitted_it() {
        // The exact shape `boundary::emit::assemble_event` builds — no
        // `clientversion` anywhere in it.
        let event = CloudEvent {
            envelope: serde_json::json!({
                "specversion": "1.0",
                "id": "evt_boundary",
                "source": "claude-code",
                "type": "ai.openlatch.economics.usage",
                "datacontenttype": "application/json",
                "data": {"tokens": 42},
            }),
            agent_id: "agt_test".to_string(),
        };

        let stamped = stamp_extensions(&event);

        assert_eq!(
            stamped["clientversion"].as_str(),
            Some(env!("OPENLATCH_VERSION")),
            "an envelope that reaches egress without a version must leave with one"
        );
        // The pre-existing stamp must survive alongside the new one.
        assert_eq!(stamped["agentid"].as_str(), Some("agt_test"));
        // And nothing the producer set may be disturbed.
        assert_eq!(stamped["data"]["tokens"], 42);
        assert_eq!(stamped["type"], "ai.openlatch.economics.usage");
    }

    /// The attribute identifies the **forwarder**, not the emitter, so the
    /// daemon's own version always wins — the same rule
    /// `daemon/handlers.rs::process_envelope` applies to hook envelopes. A
    /// stale `openlatch-hook` binary reporting an older version must not be
    /// what the platform's fleet-readiness check reads.
    #[test]
    fn clientversion_overwrites_a_stale_producer_value() {
        let event = CloudEvent {
            envelope: serde_json::json!({
                "id": "evt_stale_hook",
                "clientversion": "0.0.1-stale",
            }),
            agent_id: "agt_test".to_string(),
        };

        let stamped = stamp_extensions(&event);

        assert_eq!(
            stamped["clientversion"].as_str(),
            Some(env!("OPENLATCH_VERSION")),
            "the forwarder's version must win over the emitter's"
        );
    }

    /// Egress coverage is total: every element of every batch carries
    /// `clientversion`, including producers that never set it. Complements the
    /// unit tests above by proving it through the real worker + POST path.
    #[tokio::test]
    async fn clientversion_stamped_on_every_envelope() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(20);
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 3,
            batch_max_wait_ms: 600_000,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        // Mixed producers: unversioned (boundary/config) and a stale hook.
        for i in 0..6 {
            let envelope = if i % 2 == 0 {
                serde_json::json!({"id": format!("evt_{i}")})
            } else {
                serde_json::json!({"id": format!("evt_{i}"), "clientversion": "0.0.1-stale"})
            };
            tx.send(CloudEvent {
                envelope,
                agent_id: format!("agt_{i}"),
            })
            .await
            .unwrap();
        }
        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle).await;

        let bodies = captured.lock().unwrap().clone();
        let mut seen = 0;
        for body in &bodies {
            let elements: Vec<serde_json::Value> = serde_json::from_str(body).unwrap();
            for element in elements {
                assert_eq!(
                    element["clientversion"].as_str(),
                    Some(env!("OPENLATCH_VERSION")),
                    "every envelope on the wire must carry the forwarder version: {element}"
                );
                seen += 1;
            }
        }
        assert_eq!(seen, 6, "every enqueued event must reach the wire");
    }

    /// The status → `CloudError` mapping is unchanged by batching; only the
    /// unit it operates on went from one event to N.
    #[tokio::test]
    async fn error_mapping_unchanged() {
        let dir = tempfile::tempdir().unwrap();
        let key = SecretString::from("test-api-key".to_string());
        let batch = vec![prepare_one(serde_json::json!({"id": "evt_1"})).unwrap()];

        for (status, header, expected) in [
            (401u16, None, CloudError::AuthError),
            (403, None, CloudError::AuthError),
            (
                429,
                Some(("Retry-After", "7")),
                CloudError::RateLimit {
                    retry_after_secs: 7,
                },
            ),
            (500, None, CloudError::ServerError),
            (503, None, CloudError::ServerError),
            (418, None, CloudError::ClientError(418)),
        ] {
            let mut server = mockito::Server::new_async().await;
            let mut mock = server
                .mock("POST", "/api/v1/events/ingest")
                .with_status(status as usize);
            if let Some((name, value)) = header {
                mock = mock.with_header(name, value);
            }
            let _mock = mock.with_body("{}").create_async().await;

            let config = CloudConfig {
                api_url: server.url(),
                ..Default::default()
            };
            let client = build_cloud_client(&config);
            let err = post_batch(&client, &config, &key, &batch, dir.path())
                .await
                .expect_err("non-2xx must map to an error");
            assert_eq!(
                format!("{err:?}"),
                format!("{expected:?}"),
                "HTTP {status} must map to {expected:?}"
            );
        }

        // 429 with no parsable Retry-After falls back to the config default.
        let mut server = mockito::Server::new_async().await;
        let _mock = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(429)
            .with_body("{}")
            .create_async()
            .await;
        let config = CloudConfig {
            api_url: server.url(),
            rate_limit_default_secs: 42,
            ..Default::default()
        };
        let client = build_cloud_client(&config);
        let err = post_batch(&client, &config, &key, &batch, dir.path())
            .await
            .unwrap_err();
        assert!(
            matches!(
                err,
                CloudError::RateLimit {
                    retry_after_secs: 42
                }
            ),
            "missing Retry-After must fall back to rate_limit_default_secs, got {err:?}"
        );

        // Transport failure — nothing listening on port 1.
        let config = CloudConfig {
            api_url: "http://127.0.0.1:1".to_string(),
            timeout_connect_ms: 500,
            timeout_total_ms: 1000,
            ..Default::default()
        };
        let client = build_cloud_client(&config);
        let err = post_batch(&client, &config, &key, &batch, dir.path())
            .await
            .unwrap_err();
        assert!(
            matches!(err, CloudError::Network),
            "transport failure must map to Network, got {err:?}"
        );
    }

    /// The regression test for "extend the `select!`, don't replace it": the
    /// credential-refresh and health-probe branches must keep firing while the
    /// buffer is non-empty and the batch timer is armed. Replacing the loop
    /// with a two-branch version would silently break auth recovery, the
    /// `cloud_status` on `/metrics`, outbox-recovery notifications and
    /// emergency-mode transitions.
    #[tokio::test]
    async fn existing_select_branches_still_fire() {
        let mut server = mockito::Server::new_async().await;
        let health = server
            .mock("GET", "/api/v1/health")
            .with_status(200)
            .with_body(r#"{"status":"ok"}"#)
            .expect_at_least(1)
            .create_async()
            .await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(20);
        let provider = TestCredentialProvider::with_key("test-api-key");
        let observer = provider.clone();
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            credential_poll_interval_ms: 50,
            batch_max_events: 50,
            // Neither trigger may fire during the observation window: the
            // point is that the OTHER branches still run while the batch waits.
            batch_max_wait_ms: 600_000,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config,
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        for i in 0..3 {
            tx.send(evt(&format!("evt_{i}"))).await.unwrap();
        }
        tokio::time::sleep(std::time::Duration::from_millis(400)).await;

        assert!(
            captured.lock().unwrap().is_empty(),
            "the buffer must still be held — otherwise this proves nothing"
        );
        health.assert_async().await;
        assert!(
            observer.retrievals() >= 2,
            "the credential-poll branch must keep firing while a batch waits, got {} retrievals",
            observer.retrievals()
        );
        assert_eq!(
            state.consecutive_probe_failures(),
            0,
            "successful health probes must still be recorded"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(3), handle).await;
        assert_eq!(batch_sizes(&captured), vec![3]);
    }

    /// The guarantee `biased;` + timer-first exists to protect: with the
    /// channel continuously ready and the size trigger out of reach, the batch
    /// still leaves on the timer rather than piling up until shutdown.
    ///
    /// Scope, honestly: this asserts the *guarantee*, not the *keyword*.
    /// Deleting `biased;` does NOT fail this test — an unbiased `select!`
    /// still gives the due timer roughly even odds on each iteration, so it
    /// wins after a couple of loops and the flush is merely late by
    /// microseconds, not absent. `biased;` is what makes the bound
    /// deterministic instead of probabilistic (starvation probability decays
    /// as 0.5^k rather than being zero), which is why it stays. Do not read a
    /// green run here as licence to remove it.
    #[tokio::test]
    async fn due_timer_wins_over_saturated_channel() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(64);
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            // Out of reach: the timer is the ONLY thing that can flush.
            batch_max_events: 100_000,
            batch_max_wait_ms: 25,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        // Keep `rx` continuously ready for ~100ms — four batch deadlines.
        let producer_tx = tx.clone();
        let producer = tokio::spawn(async move {
            for i in 0..2000u32 {
                if producer_tx.send(evt(&format!("evt_{i:04}"))).await.is_err() {
                    break;
                }
                if i % 20 == 0 {
                    tokio::time::sleep(std::time::Duration::from_millis(1)).await;
                }
            }
        });
        producer.await.unwrap();

        // Read BEFORE closing the channel: the shutdown flush would otherwise
        // hand us a request the timer never produced.
        let mid_stream = captured.lock().unwrap().len();
        assert!(
            mid_stream >= 2,
            "the due timer must win against a continuously-ready channel — \
             expected multiple mid-stream flushes, got {mid_stream}"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(10), handle).await;
    }

    /// `cloud_forwarded_count` counts EVENTS, not requests. Recording one per
    /// successful batch would under-report by up to 100x on `/metrics` and in
    /// `openlatch status`.
    #[tokio::test]
    async fn forwarded_count_is_batch_length() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        let (tx, rx) = mpsc::channel::<CloudEvent>(100);
        let state = CloudState::new();
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_events: 50,
            batch_max_wait_ms: 600_000,
            ..Default::default()
        };

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
        ));

        for i in 0..50 {
            tx.send(evt(&format!("evt_{i:02}"))).await.unwrap();
        }
        tokio::time::sleep(std::time::Duration::from_millis(400)).await;

        assert_eq!(
            batch_sizes(&captured),
            vec![50],
            "one request for 50 events"
        );
        assert_eq!(
            state.forwarded_count(),
            50,
            "forwarded_count must be the batch LENGTH, not 1"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
    }

    // -----------------------------------------------------------------
    // split_batches — unit coverage for the greedy split itself
    // -----------------------------------------------------------------

    fn prepared_of(size: usize, count: usize) -> Vec<PreparedEvent> {
        (0..count)
            .map(|i| {
                prepare_one(serde_json::json!({"id": format!("evt_{i}"), "data": "x".repeat(size)}))
                    .expect("fixture must be sendable")
            })
            .collect()
    }

    #[test]
    fn split_batches_caps_on_element_count() {
        let prepared = prepared_of(8, 250);
        let groups = split_batches(&prepared, 50);
        assert_eq!(
            groups.iter().map(|g| g.len()).collect::<Vec<_>>(),
            vec![50; 5]
        );
    }

    #[test]
    fn split_batches_never_exceeds_the_platform_element_cap() {
        let prepared = prepared_of(8, 250);
        // The tuning knob may not raise the batch above the platform's cap.
        let groups = split_batches(&prepared, 100_000);
        assert!(groups.iter().all(|g| g.len() <= MAX_BATCH_EVENTS));
        assert_eq!(groups[0].len(), MAX_BATCH_EVENTS);
    }

    #[test]
    fn split_batches_closes_on_the_byte_cap() {
        // ~16KB each: 16 fit under 256KB, 17 do not.
        let prepared = prepared_of(16_000, 40);
        let groups = split_batches(&prepared, 100);
        for group in &groups {
            let framed: usize = group.iter().map(|e| e.json.len()).sum::<usize>()
                + 2
                + group.len().saturating_sub(1);
            assert!(
                framed <= MAX_BATCH_BYTES,
                "group of {} serialises to {framed} bytes",
                group.len()
            );
        }
        assert_eq!(
            groups.iter().map(|g| g.len()).sum::<usize>(),
            40,
            "the split must not lose events"
        );
        assert!(groups.len() > 1, "40 x 16KB cannot be one request");
    }

    #[test]
    fn split_batches_handles_a_single_event() {
        let prepared = prepared_of(8, 1);
        let groups = split_batches(&prepared, 50);
        assert_eq!(groups.len(), 1);
        assert_eq!(groups[0].len(), 1);
    }

    #[test]
    fn prepare_one_rejects_an_unbatchable_event() {
        assert!(
            prepare_one(serde_json::json!({"id": "big", "data": "x".repeat(300_000)})).is_none(),
            "an event that cannot fit in any batch must be dropped, not retried forever"
        );
        assert!(prepare_one(serde_json::json!({"id": "small"})).is_some());
    }
}