openlatch-client 0.5.8

//! 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::{BTreeMap, 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::CredentialProvider;
use super::{
    body_code_is_site_license, parse_retry_after, CloudConfig, CloudError, CloudEvent, CloudState,
    LICENSE_RETRY_SECS, MAX_RETRY_AFTER_SECS, SITE_LICENSE_CODES,
};
use crate::core::cloud::envelope::build_cloud_headers;
use crate::model_relay::session::{
    resolve_session_scoped, Assurance, RequestSignals, SessionRegistry,
};
use crate::model_relay::wire_format::WireFormat;

/// The code a 402 carries when its body could not be parsed, or names a code
/// this client does not recognise (D-09).
///
/// 402 is RFC-undefined, so a gateway or a proxy may answer one with anything
/// at all; retaining the batch under a generic code is the safe reading.
const LICENSE_UNKNOWN_CODE: &str = "license_unknown";

/// 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. The proxy route, the TLS
/// trust source and the connection pool (CLOUD-06) come from the egress factory; the two
/// deadlines are per-deployment, so they ride in as caller overrides.
pub fn build_cloud_client(
    config: &CloudConfig,
    egress: &crate::egress::EgressConfig,
) -> Result<reqwest::Client, crate::error::OlError> {
    crate::egress::build_client_with(
        crate::egress::Consumer::CloudWorker,
        egress,
        crate::egress::Timeouts {
            connect: Some(Duration::from_millis(config.timeout_connect_ms)),
            total: Some(Duration::from_millis(config.timeout_total_ms)),
        },
    )
}

/// 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
/// - `egress`: the resolved proxy route every POST takes, and the runtime
///   egress health its outcomes are reported into
/// - `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.
#[allow(clippy::too_many_arguments)]
pub async fn run_cloud_worker(
    rx: mpsc::Receiver<CloudEvent>,
    credential_provider: Arc<dyn CredentialProvider>,
    config: CloudConfig,
    egress: crate::egress::EgressReporter,
    cloud_state: CloudState,
    openlatch_dir: PathBuf,
    outbox: Option<Arc<Outbox>>,
    shutdown: Option<tokio::sync::watch::Receiver<bool>>,
    source_formats: SourceFormats,
    sessions: Arc<SessionRegistry>,
) {
    let mut rx = rx;
    // A client we cannot build is a configuration fault, not a transient one: retrying it
    // would re-fail identically every restart, so say why once and stop.
    let client = match build_cloud_client(&config, egress.config()) {
        Ok(c) => crate::egress::ClientHandle::of(c),
        Err(e) => {
            tracing::error!(code = %e.code, error = %e.message, "cloud worker http client init failed; cloud forwarding is off");
            return;
        }
    };
    run_cloud_worker_on(
        &mut rx,
        credential_provider,
        config,
        egress,
        cloud_state,
        openlatch_dir,
        outbox,
        shutdown,
        client,
        source_formats,
        sessions,
    )
    .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,
    egress: crate::egress::EgressReporter,
    cloud_state: CloudState,
    openlatch_dir: PathBuf,
    outbox: Option<Arc<Outbox>>,
    shutdown: Option<tokio::sync::watch::Receiver<bool>>,
    client: crate::egress::ClientHandle,
    source_formats: SourceFormats,
    sessions: Arc<SessionRegistry>,
) {
    // 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 credential_poll_interval = Duration::from_millis(config.credential_poll_interval_ms);

    // NOTE: the durable-outbox drain no longer lives here. It used to be a
    // bare `tokio::spawn` nested in this function — a panic in it was
    // invisible (nothing joined the handle) and permanent (this function
    // only spawns it once, at startup, and the outer cloud worker itself
    // does not panic just because its fire-and-forget child did). It is now
    // [`run_outbox_drain_on`], registered as its OWN supervised subsystem
    // beside the cloud worker in `daemon::mod` (issue #306, contract 3).

    // 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 system 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,
        egress: &egress,
        formats: &source_formats,
        sessions: &sessions,
        credential_provider: &credential_provider,
    };

    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.
            //
            // `biased;` lists this branch ahead of `rx.recv()` below, so a
            // caller that sends events and signals shutdown back-to-back can
            // have this branch win while those events are still sitting,
            // received-but-unread, in the channel — the same iteration never
            // gets a chance to run the `rx.recv()` arm that would have
            // buffered them. Draining here closes that gap: everything
            // already queued at this instant gets the same admission checks
            // the live path applies, then rides the post-loop flush below
            // (issue #380).
            //
            // The drain is bounded by a snapshot of `rx.len()`, not an
            // unbounded `while let Ok(...) = rx.try_recv()`. `try_recv`
            // itself never blocks, but a producer on another thread that
            // keeps refilling the channel as fast as this drains it could
            // otherwise keep the loop finding "just one more" event forever
            // — bounding it to what was already queued the moment shutdown
            // fired means the loop always terminates in exactly `queued`
            // steps regardless of what arrives afterward. `flush_batch`'s
            // own wire-size caps are the bound on how large that queued
            // snapshot may safely be; the live arm's size-trigger flush is
            // for steady-state batching, not shutdown, so it has no
            // equivalent role here.
            _ = shutdown_rx.changed() => {
                tracing::info!("cloud worker: shutdown signalled, flushing in-flight batch");
                let queued = rx.len();
                for _ in 0..queued {
                    let Ok(event) = rx.try_recv() else { break };
                    if auth_error {
                        spool_event(
                            outbox.as_ref(),
                            &event,
                            &source_formats,
                            &sessions,
                            config.host_id.as_deref(),
                            SpoolReason::AuthError,
                        );
                        continue;
                    }
                    if current_key.is_none() {
                        spool_event(
                            outbox.as_ref(),
                            &event,
                            &source_formats,
                            &sessions,
                            config.host_id.as_deref(),
                            SpoolReason::NoCredential,
                        );
                        continue;
                    }
                    buf.push(event);
                }
                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();
                // ONLY the probe call is gated, never the whole branch: the
                // emergency-mode detector tick below rides this same cadence and
                // has to keep running exactly while traffic is flowing.
                // THE LOAD POINT — once per probe. Probing through the client the
                // daemon built at start-up would keep asking a proxy that is already gone.
                if health_probe_due(&egress) {
                    if let Some(probe_client) = client.current() {
                        run_health_probe(
                            &probe_client,
                            &config,
                            &cloud_state,
                            &egress,
                            &mut degraded,
                        )
                        .await;
                    }
                }
                // 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,
                        &source_formats,
                        &sessions,
                        config.host_id.as_deref(),
                        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 system 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,
                        &source_formats,
                        &sessions,
                        config.host_id.as_deref(),
                        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,
                flush_ctx.formats,
                flush_ctx.sessions,
                flush_ctx.config.host_id.as_deref(),
                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> {
    /// The swap handle, re-read once per flush — never a captured client. See
    /// [`crate::egress::EgressClients`]: the daemon builds this worker's client once, so
    /// without the re-read a healed route would never reach the highest-volume consumer
    /// on the host.
    client: &'a crate::egress::ClientHandle,
    config: &'a CloudConfig,
    openlatch_dir: &'a Path,
    outbox: Option<&'a Arc<Outbox>>,
    cloud_state: &'a CloudState,
    /// Where this worker's transport outcomes are reported. The cloud worker is
    /// the client's highest-volume egress consumer, so it is the site that
    /// notices a dead proxy first on a machine that is actually being used.
    egress: &'a crate::egress::EgressReporter,
    /// `source` → wire format, for the `wireformat` stamp. Built once at worker
    /// start; see [`SourceFormats`].
    formats: &'a SourceFormats,
    /// The live-session registry, for the `subject` stamp. The same `Arc` the
    /// hook path writes and the model relay reads; see [`stamp_session`].
    sessions: &'a SessionRegistry,
    /// Invalidated exactly once per latch onset (see the `AuthError` arm
    /// below) so the credential-poll loop's next `retrieve()` re-reads the
    /// backing store instead of a memoized answer — otherwise a rotated
    /// keychain entry never becomes visible and the latch never clears.
    credential_provider: &'a Arc<dyn CredentialProvider>,
}

/// Meet a licence refusal: set the gate, and **keep the events**.
///
/// One helper for every arm that can meet one — the live POST, each of the
/// three retries, and the pre-group guard — because the rule they share is the
/// invariant: a refusal is never a credential failure, never an
/// `OUTBOX_MAX_ATTEMPTS` charge, and never a reason to drop a batch that has
/// somewhere to wait.
///
/// `record_drops` fires **only** when there is no outbox, because then the
/// events really are gone; with one they are spooled and the counter would
/// double-count a batch that is still on disk.
fn gate_and_spool(
    ctx: &FlushCtx<'_>,
    group: &[PreparedEvent],
    code: String,
    licensing_url: Option<String>,
    retry_after: Option<Duration>,
) {
    let entering = ctx.cloud_state.license_gate().is_none();
    let retry_in_secs = retry_after
        .unwrap_or(Duration::from_secs(LICENSE_RETRY_SECS))
        .as_secs();
    ctx.cloud_state
        .set_license_gate(code.clone(), licensing_url.clone(), retry_after);

    // WARN once on entering, DEBUG for every spool after — the `degraded`
    // streak idiom, for the same reason: an expired contract must not produce
    // one WARN per batch for as long as it stays expired.
    if entering {
        tracing::warn!(
            code = crate::error::ERR_CLOUD_LICENSE_BLOCKED,
            license_code = %code,
            // `Option<&str>` records the bare URL when present and nothing
            // when absent — never `Some("…")`.
            licensing_url = licensing_url.as_deref(),
            retry_in_secs,
            batch_size = group.len(),
            "cloud worker: the platform refused ingest on licensing grounds — events are queued, enforcement continues"
        );
    } else {
        tracing::debug!(
            code = crate::error::ERR_CLOUD_LICENSE_BLOCKED,
            license_code = %code,
            "cloud worker: still license-blocked — queueing"
        );
    }

    if ctx.outbox.is_some() {
        spool_group(ctx.outbox, group, SpoolReason::LicenseRefused);
    } else {
        ctx.cloud_state.record_drops(group.len() as u64);
    }
}

/// Note that a POST has just succeeded while a licence gate was standing.
///
/// Shared by the live path and the outbox drain: whichever reaches the platform
/// first is proof the refusal is over, and the recovery is logged exactly once.
fn clear_license_gate_on_success(cloud_state: &CloudState) {
    if cloud_state.clear_license_gate() {
        tracing::warn!("cloud worker: license gate cleared — forwarding resumed");
    }
}

/// 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,
        ctx.formats,
        ctx.sessions,
        ctx.config.host_id.as_deref(),
    );
    if prepared.is_empty() {
        return;
    }

    // THE LOAD POINT — once per flush, before the first POST. A batch is the natural
    // model relay: every group below goes out on one coherent route, and the next flush picks
    // up whatever the self-heal pass installed in the meantime.
    let Some(client) = ctx.client.current() else {
        // No route is permitted (`[proxy] allow_direct = false` with nothing working).
        // Spooling is the honest answer: the events survive on disk and the drain replays
        // them the moment a route exists. Sending them direct is the one thing that must
        // never happen here.
        tracing::debug!(
            code = crate::error::ERR_DIRECT_FORBIDDEN,
            batch_size = prepared.len(),
            "cloud worker: no egress route is permitted; spooling instead of forwarding"
        );
        spool_group(ctx.outbox, &prepared, SpoolReason::Network);
        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;
        }
        // The licence cadence, checked before the POST rather than after it:
        // while `until` is in the future the answer is already known. The gate
        // is not re-set here — `set_license_gate` would push `until` forward on
        // every flush and the cadence would never elapse.
        if ctx.cloud_state.license_gate_active() {
            if ctx.outbox.is_some() {
                spool_group(ctx.outbox, group, SpoolReason::LicenseRefused);
            } else {
                ctx.cloud_state.record_drops(group.len() as u64);
            }
            continue;
        }
        let n = group.len() as u64;
        match post_batch(
            &client,
            ctx.config,
            key,
            group,
            ctx.openlatch_dir,
            ctx.egress,
        )
        .await
        {
            Ok(()) => {
                note_recovery(degraded, ctx.cloud_state);
                clear_license_gate_on_success(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"
                );
                // This arm only runs on the false->true transition: every
                // OTHER group in this flush, and every flush while the latch
                // is already up, takes the early `if *auth_error` branch
                // above instead. So this fires exactly once per latch —
                // dropping the memo here, not on every poll, is what keeps a
                // permanently-revoked key from re-triggering a keychain read
                // (and macOS authorization dialog) every 60 seconds.
                //
                // The credential just proved wrong is also the one the
                // process-wide keyring memo is holding onto (see
                // `core::auth::keyring::READ_MEMO`): without this, the next
                // credential-poll `retrieve()` would hand back the SAME
                // memoized secret, `key_changed` would read false forever,
                // and a real rotation sitting in the keychain would never
                // become visible short of a daemon restart.
                ctx.credential_provider.invalidate();
                *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(
                    &client,
                    ctx.config,
                    key,
                    group,
                    ctx.openlatch_dir,
                    ctx.egress,
                )
                .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),
                    // A refusal met on the retry is still a refusal — the
                    // batch waits out the cadence rather than being dropped.
                    Err(CloudError::LicenseRefused {
                        code,
                        licensing_url,
                        retry_after,
                    }) => gate_and_spool(ctx, group, code, licensing_url, retry_after),
                    Err(_) => {
                        ctx.cloud_state.record_drops(n);
                        spool_group(ctx.outbox, group, SpoolReason::RateLimit);
                    }
                }
            }
            // No `record_drops` while an outbox exists, no `auth_error`, and
            // no retry inside the arm: the cadence is the retry.
            Err(CloudError::LicenseRefused {
                code,
                licensing_url,
                retry_after,
            }) => gate_and_spool(ctx, group, code, licensing_url, retry_after),
            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(
                    &client,
                    ctx.config,
                    key,
                    group,
                    ctx.openlatch_dir,
                    ctx.egress,
                )
                .await
                {
                    Ok(()) => ctx.cloud_state.record_successful_forwards(n),
                    // A refusal met on the retry is still a refusal — the
                    // batch waits out the cadence rather than being dropped.
                    Err(CloudError::LicenseRefused {
                        code,
                        licensing_url,
                        retry_after,
                    }) => gate_and_spool(ctx, group, code, licensing_url, retry_after),
                    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(
                    &client,
                    ctx.config,
                    key,
                    group,
                    ctx.openlatch_dir,
                    ctx.egress,
                )
                .await
                {
                    Ok(()) => ctx.cloud_state.record_successful_forwards(n),
                    // A refusal met on the retry is still a refusal — the
                    // batch waits out the cadence rather than being dropped.
                    Err(CloudError::LicenseRefused {
                        code,
                        licensing_url,
                        retry_after,
                    }) => gate_and_spool(ctx, group, code, licensing_url, retry_after),
                    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);
            }
            Err(CloudError::CompatibilityUnavailable) => {
                tracing::warn!(
                    batch_size = n,
                    "cloud worker: decision-event compatibility unavailable — retaining events for renegotiation"
                );
                spool_group(ctx.outbox, group, SpoolReason::Compatibility);
                ctx.cloud_state.policy_refresh_notify.notify_one();
            }
        }
    }
}

/// The unauthenticated liveness URL this worker probes.
fn health_url(config: &CloudConfig) -> String {
    format!("{}/api/v1/health", config.api_url.trim_end_matches('/'))
}

/// 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 resp = client.get(health_url(config)).send().await?;
    resp.error_for_status().map(|_| ())
}

/// Whether the idle probe is worth issuing on this tick.
///
/// The probe exists to break **silence**. Recorded outcomes from real traffic
/// answer the same question for free and cost nothing extra, so a host that is
/// actually being used never pays for a probe — that is the invariant this
/// function is the whole implementation of.
///
/// A route that reports nowhere has no silence to measure, so it probes on every
/// tick exactly as it did before this existed. That is what keeps the worker's
/// own test suite, which drives the probe deliberately, unchanged.
fn health_probe_due(egress: &crate::egress::EgressReporter) -> bool {
    egress.state().is_none_or(|state| state.is_idle())
}

/// Run the idle probe, then — only on failure — exactly one immediate retry.
///
/// The retry is what bounds the visibility budget. `failed` needs two
/// consecutive failures, so without an immediate second attempt a proxy that
/// died on an idle host would need two whole cadence intervals to surface.
/// With it the worst case is one silence window plus two probe timeouts.
///
/// The retry lives **here**, with whoever owns the probe. A monitor-issued retry
/// would be a second probe path against the same endpoint.
async fn run_health_probe(
    client: &reqwest::Client,
    config: &CloudConfig,
    cloud_state: &CloudState,
    egress: &crate::egress::EgressReporter,
    degraded: &mut Option<&'static str>,
) {
    // Cleared on drop, so an early return cannot leave `/admin/egress/status`
    // claiming a probe is still in flight.
    let _probing = egress.state().map(|state| state.probe_guard());
    let url = health_url(config);

    for attempt in 0..2u8 {
        match cloud_health_check(client, config).await {
            Ok(()) => {
                note_recovery(degraded, cloud_state);
                cloud_state.record_health_ok();
                egress.record_ok(&url);
                // 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();
                return;
            }
            Err(e) => {
                cloud_state.record_probe_failure();
                egress.record_failure(&url, &e);
                tracing::debug!(
                    error = %e,
                    attempt = attempt + 1,
                    consecutive_failures = cloud_state.consecutive_probe_failures(),
                    "cloud health check failed"
                );
            }
        }
    }
}

/// 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,
    LicenseRefused,
    Compatibility,
}

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",
            SpoolReason::LicenseRefused => "license_refused",
            SpoolReason::Compatibility => "compatibility",
        }
    }
}

/// 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,
    formats: &SourceFormats,
    sessions: &SessionRegistry,
    host_id: Option<&str>,
    reason: SpoolReason,
) {
    if outbox.is_none() {
        return;
    }
    spool_envelope(
        outbox,
        &stamp_extensions(event, formats, sessions, host_id),
        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);
    }
}

/// `source` (the CloudEvents agent name) → the wire format that agent's request
/// plane speaks.
///
/// Built **once**, by the daemon, from `detect_agents()`, and handed in. Two
/// reasons it is a parameter rather than a global or a per-event lookup:
///
/// - the worker flushes batches, so a `detect_agents()` call per envelope would
///   stat the filesystem on every event;
/// - a global cannot be varied per test, and this module's own tests have to
///   assert `codex-cli → openai-responses` on a developer machine where
///   `detect_agents()` finds no Codex at all.
///
/// A host that installs a new agent mid-run gets the attribute after the next
/// daemon restart. That is acceptable, and it is said out loud here rather than
/// discovered.
///
/// `core::cloud` never calls `crate::hooks::` — that would invert the
/// `core` → `hooks` direction. The daemon builds the map and passes it down.
///
/// # Why a shared handle rather than a plain map
///
/// It used to be a `BTreeMap` moved into the worker, and "after the next daemon
/// restart" was the whole story. That was true while every shipped binding
/// answered a CONSTANT format. It stopped being true the moment a GUI-hosted
/// agent's format became a function of which provider the customer is on: they
/// switch provider in the UI, the daemon re-wires them on its next tick, and
/// every envelope that agent produces from then until a restart carries the
/// format of the provider they LEFT — stamped on hook, config and economics
/// events alike, with nothing to say it is stale.
///
/// So the map is shared and the daemon updates it at the one place the format
/// is known to have changed: where the wiring is written. Reads stay cheap — a
/// `BTreeMap` lookup under a read lock, per envelope batch, not per event — and
/// the `core` → `hooks` direction is untouched, because the daemon still does
/// all the detecting.
#[derive(Debug, Clone, Default)]
pub struct SourceFormats(Arc<std::sync::RwLock<BTreeMap<&'static str, WireFormat>>>);

impl SourceFormats {
    /// An empty map. Every test's starting point, and a daemon's before
    /// detection has run.
    pub fn new() -> Self {
        Self::default()
    }

    /// The format recorded for `source`, if any.
    ///
    /// Poison-tolerant: a panicked writer has already failed, and losing the
    /// `wireformat` attribute on every subsequent event — or taking the egress
    /// worker down with an unwrap — is a worse outcome than reading the value
    /// it was holding.
    pub fn get(&self, source: &str) -> Option<WireFormat> {
        match self.0.read() {
            Ok(map) => map.get(source).copied(),
            Err(poisoned) => poisoned.into_inner().get(source).copied(),
        }
    }

    /// Record the format `source` now speaks, replacing any earlier answer.
    ///
    /// Called from the daemon's wiring path, which is the single place that
    /// learns a plane's format changed.
    pub fn set(&self, source: &'static str, format: WireFormat) {
        match self.0.write() {
            Ok(mut map) => {
                map.insert(source, format);
            }
            Err(poisoned) => {
                poisoned.into_inner().insert(source, format);
            }
        }
    }
}

impl FromIterator<(&'static str, WireFormat)> for SourceFormats {
    fn from_iter<T: IntoIterator<Item = (&'static str, WireFormat)>>(iter: T) -> Self {
        Self(Arc::new(std::sync::RwLock::new(iter.into_iter().collect())))
    }
}

/// The wire format for an envelope's `source`, or `None` when the source names
/// no agent with a request plane on this host.
fn wire_format_for_source(
    envelope: &serde_json::Value,
    formats: &SourceFormats,
) -> Option<WireFormat> {
    let source = envelope.get("source")?.as_str()?;
    formats.get(source)
}

/// 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 — `model_relay/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,
    formats: &SourceFormats,
    sessions: &SessionRegistry,
    host_id: Option<&str>,
) -> 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()),
        );
        // `hostid` (I-4), filled ONLY into an envelope that carries none.
        //
        // `daemon/handlers.rs` stamps it on hook envelopes; the config
        // monitor, the tamper reconciler and the model relay build their own
        // and never pass through there. Without this fill the platform keys
        // those `agent:<agent_id>` and counts a second licensed host beside
        // the machine's real one — so a laptop with a config monitor would
        // arrive as two hosts. Licensing counts the machine for every
        // producer, which is why this one attribute reaches the egress while
        // the `ol*` extensions deliberately do not.
        //
        // `or_insert_with`, never `insert`: a hook envelope keeps exactly what
        // `host_stamp` put there, and a producer that sets its own is never
        // overwritten. `hostname` is NOT filled here — it is an identity
        // signal, Live and capture only, and stays on the hook path.
        if let Some(host_id) = host_id {
            obj.entry("hostid")
                .or_insert_with(|| serde_json::Value::String(host_id.to_string()));
        }
        // OPENLATCH_VERSION, not CARGO_PKG_VERSION — see build.rs.
        obj.insert(
            "clientversion".to_string(),
            serde_json::Value::String(env!("OPENLATCH_VERSION").to_string()),
        );
        // The wire format is the same class of thing as `agentid`: a standing
        // attribute of the AGENT, true whether the hook, the model relay or the
        // config monitor observed it — so it is stamped here, not per builder.
        // (`handlers.rs`'s `proxytype`/`proxysource`/`proxyintercepted` are
        // facts about the client's ROUTE to the platform and stay partial on
        // purpose; that precedent does not reach this attribute.)
        //
        // ALWAYS PRESENT, carrying `"unknown"` when the source cannot be
        // resolved — the source is `"unknown"`, the agent has no binding on this
        // host, or its binding declares no request plane. The platform gets a
        // column it can group and join on with no null handling, and `"unknown"`
        // is one of the declared format names, so it claims no protocol. It is
        // the same shape `ai.openlatch.session.source` already uses on the very
        // same row.
        //
        // A producer that observed the format itself wins, and exactly one
        // does: the model relay saw the request and knows which protocol it
        // spoke. The map is per SOURCE, so it can only answer for an agent
        // wired to one request plane — an agent carrying many providers at once
        // (Cline) has no entry at all, and its relay rows would be stamped
        // `unknown` while their own `data` named the provider they went to.
        let producer_named_it = obj
            .get("wireformat")
            .and_then(|value| value.as_str())
            .is_some_and(|format| !format.is_empty());
        if !producer_named_it {
            let fmt =
                wire_format_for_source(&event.envelope, formats).unwrap_or(WireFormat::Unknown);
            obj.insert(
                "wireformat".to_string(),
                serde_json::Value::String(fmt.as_str().to_string()),
            );
        }
        stamp_session(obj, &event.agent_id, sessions);
    }
    envelope
}

/// Give an envelope that named no session one, and say how sure we are.
///
/// The platform's whole UI is session-scoped — every event is shown relative to
/// a session, which is relative to an agent — so an event arriving with no
/// `subject` cannot be displayed at all. The hook path and the model relay path
/// both name their session; the config monitor and the tamper reconciler have
/// none to name, and their events were invisible because of it. This is where
/// they get one, for the same reason `clientversion` is stamped here: those
/// producers hand-roll their envelopes, and the registry that holds the answer
/// is not reachable from any of them.
///
/// The honest objection is that a config change or a tamper detection is **not
/// necessarily caused by a session** — a developer can edit `settings.json`
/// with no agent running at all. So the correlation is filled in *and* labelled:
/// [`Assurance`] rides along in `data` under the same key the model relay already
/// writes (`emit.rs`), so the platform reads one key everywhere and an
/// `inferred` or `unknown` session is visibly a correlation rather than a fact.
///
/// Three rules, in order:
///
/// - **A subject the producer set wins, untouched, `data` included.** The
///   producer knew: a hook envelope carries the agent's own `session_id`, and a
///   model relay envelope carries the cascade's answer *with* its own assurance
///   already in `data`. Relabelling an `inferred` model relay row as `attested`
///   would be a lie, and writing into a hook envelope's `data` would rewrite the
///   agent's raw payload, which the perimeter forbids outright.
/// - **No subject, a live session** → that session id, carrying whatever
///   assurance the registry returned (`attested` for exactly one live session,
///   `inferred` when it had to pick the most-recently-active of several).
/// - **No subject, no live session** → the agent id, `unknown`. A stable
///   per-install bucket, deliberately chosen over a null that would hide the
///   event: `unknown` says plainly that no session was claimed. It is one bucket
///   per install — the platform can detect `subject == agentid` and render it as
///   host-level — where the `entry_id` this replaces minted a fresh phantom
///   session per hook registration.
///
/// Inserting into `data` is only ever safe because of the first rule: the branch
/// that writes runs exclusively for envelopes whose `data` we author ourselves.
fn stamp_session(
    obj: &mut serde_json::Map<String, serde_json::Value>,
    agent_id: &str,
    sessions: &SessionRegistry,
) {
    let has_subject = obj
        .get("subject")
        .and_then(|v| v.as_str())
        .is_some_and(|s| !s.is_empty());
    if has_subject {
        return;
    }

    // Scoped to THIS envelope's agent. `agent_id` is the install, which every
    // agent on the host shares, so resolving by it alone let the most-recently
    // active session win whatever produced the event — and a Claude Code
    // config snapshot landed inside a live Cline conversation, labelled
    // `inferred`, on a platform whose sessions are agent-scoped. The relay path
    // has always scoped its resolution for exactly this reason; this is the
    // same call with the same argument, from the other producer.
    //
    // No source (an envelope that named none) scopes to nothing and keeps the
    // old behaviour: a correlation across agents is still better than an event
    // the UI cannot display, and the assurance says which it was.
    let source = obj.get("source").and_then(serde_json::Value::as_str);
    let resolved = resolve_session_scoped(sessions, agent_id, &RequestSignals::default(), source);
    // `session_id: None` always comes back with `Unknown`, but read the pair
    // together rather than trusting that from a distance: the fallback subject
    // is the install, not a session, and must never be labelled as one.
    let (subject, assurance) = match resolved.session_id {
        Some(session_id) => (session_id, resolved.assurance),
        None => (agent_id.to_string(), Assurance::Unknown),
    };
    obj.insert("subject".to_string(), serde_json::Value::String(subject));

    // `data` is absent on no producer today, but `config.removed` sets it to
    // `Value::Null` — so create the object in both cases, and leave a `data`
    // that is neither (an array, a scalar) exactly as it is rather than
    // discarding a payload to make room for an annotation.
    let data = obj
        .entry("data")
        .or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
    if data.is_null() {
        *data = serde_json::Value::Object(serde_json::Map::new());
    }
    if let Some(map) = data.as_object_mut() {
        map.insert(
            crate::model_relay::session::ASSURANCE_KEY.to_string(),
            serde_json::Value::String(assurance.as_str().to_string()),
        );
    }
}

/// 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,
    formats: &SourceFormats,
    sessions: &SessionRegistry,
    host_id: Option<&str>,
) -> Vec<PreparedEvent> {
    let mut out = Vec::with_capacity(events.len());
    for event in events {
        match prepare_one(stamp_extensions(event, formats, sessions, host_id)) {
            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
}

/// The durable-outbox drain loop: one immediate pass at startup (in case a
/// previous daemon run left entries behind), then one pass per
/// `cloud_state.outbox_drain_notify` signal, forever.
///
/// Runs under the daemon's task supervisor (`core::supervision::task`),
/// registered as its own subsystem beside the cloud worker in
/// `daemon::mod` — **not** nested inside [`run_cloud_worker_on`]. It used to
/// be a bare `tokio::spawn` in there, which meant a panic mid-drain was
/// invisible (nothing joined the handle) and permanent (the enclosing
/// function spawns this task exactly once, at its own startup, and does not
/// itself panic just because a fire-and-forget child did) — the cloud worker
/// kept reporting healthy while outbox replay silently died for the rest of
/// the process lifetime (issue #306, contract 3).
///
/// The factory in `daemon::mod` calls this fresh on every restart per the
/// supervisor's invariant; every argument here is `Clone` and owned, so a
/// panic mid-pass loses nothing but that one pass — `attempts` (the
/// per-id retry counter) is rebuilt from empty, matching a restart's own
/// "quarantine budget resets per daemon lifetime" semantics.
///
/// `shutdown` is the same daemon-wide signal the cloud worker and model relay
/// take (`.claude/rules/subsystem-supervision.md`): the loop selects on it
/// alongside the drain notify and returns as soon as it fires, so the
/// supervisor's own shutdown wait (bounded by `SHUTDOWN_DRAIN`) joins this
/// task promptly instead of timing out and force-aborting it.
#[allow(clippy::too_many_arguments)]
pub async fn run_outbox_drain_on(
    outbox: Arc<Outbox>,
    // The HANDLE, not a client: the drain task outlives any one route, and a
    // self-heal swap has to reach it (a host that spent the night offline
    // drains through whatever proxy is live when it wakes up).
    client: crate::egress::ClientHandle,
    config: CloudConfig,
    credential_provider: Arc<dyn CredentialProvider>,
    cloud_state: CloudState,
    openlatch_dir: PathBuf,
    source_formats: SourceFormats,
    sessions: Arc<SessionRegistry>,
    egress: crate::egress::EgressReporter,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
    // Test-only failpoint: instance-owned, never a process-global static, so
    // one test's armed panic cannot consume itself against a different
    // test's drain task or crash a neighbour running in parallel (issue
    // #306 review, contract 3). Absent entirely outside `cfg(test)` builds.
    #[cfg(test)] fail_once: Option<Arc<std::sync::atomic::AtomicBool>>,
) {
    // Cached once per run: 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(&openlatch_dir).unwrap_or_default();
    // Volatile per-id attempt counts. A restart wipes the map and the
    // quarantine budget resets per run. 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 notify = cloud_state.outbox_drain_notify.clone();
    let ctx = OutboxDrainCtx {
        outbox: &outbox,
        client: &client,
        config: &config,
        credential_provider: &credential_provider,
        cloud_state: &cloud_state,
        openlatch_dir: &openlatch_dir,
        agent_id: &agent_id,
        attempts: &attempts,
        egress: &egress,
        formats: &source_formats,
        sessions: &sessions,
        #[cfg(test)]
        fail_once: fail_once.as_deref(),
    };

    maybe_panic_for_test(&ctx);
    drain_outbox_once(&ctx).await;
    loop {
        // `biased;` with the shutdown arm first: without it, `select!` polls
        // in a randomized order, so a notify permit already stored when
        // shutdown fires (e.g. a health-recovery notify racing `openlatch
        // stop`) wins on some fraction of runs and starts one more drain
        // pass after shutdown was already signalled. Shutdown must always
        // win when both are ready — see
        // `outbox_drain_shutdown_wins_a_race_against_notify`.
        tokio::select! {
            biased;

            _ = shutdown.wait_for(|stop| *stop) => {
                tracing::info!("outbox drain: shutdown signalled, stopping");
                return;
            }
            _ = notify.notified() => {}
        }
        // 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 cloud_state.is_emergency_mode() {
            tracing::debug!("outbox drain: skipping pass — cloud channel in emergency mode");
            continue;
        }

        // Same shape, different reason: while the licence gate holds,
        // replaying the backlog would only re-earn the refusal. Every
        // successful health probe fires `outbox_drain_notify`, so without
        // this an idle host with a backlog POSTs into the gate about once a
        // minute for the life of an expired contract. The startup drain
        // above runs before any gate can exist — the gate is in memory and
        // resets on restart.
        if cloud_state.license_gate_active() {
            tracing::debug!("outbox drain: skipping pass — license gate active");
            continue;
        }
        maybe_panic_for_test(&ctx);
        drain_outbox_once(&ctx).await;
    }
}

/// Test-only fail point: when `ctx.fail_once` is armed (`Some` and `true`),
/// makes this [`run_outbox_drain_on`] pass panic once, so a test can prove a
/// panic inside this task is observed and recovered by the daemon's task
/// supervisor rather than silently ending outbox replay for the rest of the
/// process lifetime (issue #306, contract 3). The flag is owned by whichever
/// test armed it — never a process-global static, so a parallel test's drain
/// task cannot consume or trip it. Compiled only under `cfg(test)` — no
/// footprint in a release binary.
#[cfg(test)]
fn maybe_panic_for_test(ctx: &OutboxDrainCtx<'_>) {
    if let Some(flag) = ctx.fail_once {
        if flag.swap(false, std::sync::atomic::Ordering::SeqCst) {
            panic!("injected outbox-drain failure (test)");
        }
    }
}

#[cfg(not(test))]
#[inline(always)]
fn maybe_panic_for_test(_ctx: &OutboxDrainCtx<'_>) {}

/// 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>,
    /// The swap handle, re-read once per drain pass. Same reason as [`FlushCtx::client`].
    client: &'a crate::egress::ClientHandle,
    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>>>,
    /// Replay traffic takes the same route as live traffic, so its outcomes
    /// belong in the same health picture — on a quiet host that spent the night
    /// offline, the drain is the first thing to discover the proxy is back.
    egress: &'a crate::egress::EgressReporter,
    /// `source` → wire format, so a replayed envelope leaves stamped exactly as
    /// a live one does.
    formats: &'a SourceFormats,
    /// The live-session registry, for the same reason. Re-stamping a replayed
    /// envelope is a no-op for `subject`: it was stamped before it was spooled,
    /// and a subject already present is never overwritten.
    sessions: &'a Arc<SessionRegistry>,
    /// Test-only failpoint, owned by the instance under test — see
    /// [`maybe_panic_for_test`]. Absent entirely outside `cfg(test)` builds.
    #[cfg(test)]
    fail_once: Option<&'a std::sync::atomic::AtomicBool>,
}

/// 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 system 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 `outbox_drain_notify` again the next time a health
        // probe succeeds after a credential lands.
        return;
    };

    // THE LOAD POINT — once per drain pass. The drain task is spawned once and lives for
    // the whole daemon, so without this a replay would keep using the route that was live
    // at start-up; on a host that spent the night offline that is precisely the dead one.
    let Some(drain_client) = ctx.client.current() else {
        tracing::debug!(
            code = crate::error::ERR_DIRECT_FORBIDDEN,
            "outbox drain: no egress route is permitted; entries stay queued"
        );
        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 = drain_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();
            let drain_egress = ctx.egress.clone();
            let formats = ctx.formats.clone();
            let sessions = ctx.sessions.clone();
            // The drain is the OTHER path that can meet a licence refusal, and
            // the other path that can prove one is over.
            let cloud_state = ctx.cloud_state.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, mut 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),
                    );
                    // Local retry metadata stays additive on disk so rollback
                    // readers still see a normal CloudEvent. It is removed at
                    // the shared send boundary and never becomes event data.
                    super::outbox::strip_retry_metadata(&mut envelope);
                    // 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,
                        &formats,
                        &sessions,
                        config.host_id.as_deref(),
                    )) {
                        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, &drain_egress)
                        .await
                    {
                        // A drain that replays the backlog after a renewal is
                        // a successful POST, and it is the only one an idle
                        // host will make. Without clearing here, `/metrics`
                        // would keep reporting `license_blocked`, the doctor's
                        // Cloud section would stay Degraded and `status` would
                        // keep exiting 7 until some live hook event happened to
                        // be forwarded.
                        Ok(()) => clear_license_gate_on_success(&cloud_state),
                        // 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) => {
                            // A refusal met here sets the gate exactly as the
                            // live path does, so the notify loop above stops
                            // replaying into it. The entries are kept either
                            // way — this only decides whether a gate stands.
                            if let CloudError::LicenseRefused {
                                code,
                                licensing_url,
                                retry_after,
                            } = &e
                            {
                                cloud_state.set_license_gate(
                                    code.clone(),
                                    licensing_url.clone(),
                                    *retry_after,
                                );
                            }
                            failure = Some(e);
                            break;
                        }
                    }
                }

                // A licence refusal must never spend an entry's attempt
                // budget: `OUTBOX_MAX_ATTEMPTS` exists to get the queue past a
                // permanently-rejecting envelope, and these entries are fine —
                // it is the contract that is not. Charging them would
                // quarantine the whole backlog over a renewal.
                if matches!(failure, Some(CloudError::LicenseRefused { .. })) {
                    tracing::debug!(
                        "outbox drain: halted on a license refusal — entries retained, no attempt charged"
                    );
                    return Err(());
                }

                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);
                };

                // A peer deployment or acknowledgement resolves compatibility;
                // repeatedly charging the same retained event as poison would
                // quarantine evidence before it can be renegotiated.
                if matches!(e, CloudError::CompatibilityUnavailable) {
                    ctx.cloud_state.policy_refresh_notify.notify_one();
                    tracing::debug!(
                        batch_size = group_len,
                        "outbox drain: compatibility unavailable — retaining entries without charging attempts"
                    );
                    return Err(());
                }

                // 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.expired > 0 {
                tracing::warn!(
                    code = crate::error::ERR_OUTBOX_EXPIRED,
                    expired = stats.expired,
                    remaining = ctx.outbox.pending_count(),
                    "outbox: terminally expired retained compatibility evidence"
                );
            }
            if stats.drained > 0
                || stats.failed > 0
                || stats.corrupt > 0
                || stats.quarantined > 0
                || stats.expired > 0
            {
                tracing::info!(
                    drained = stats.drained,
                    failed = stats.failed,
                    corrupt = stats.corrupt,
                    quarantined = stats.quarantined,
                    expired = stats.expired,
                    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,
    egress: &crate::egress::EgressReporter,
) -> Result<(), CloudError> {
    let mut selected_version = crate::core::protocol::contracts::read_compatibility(openlatch_dir)
        .ok()
        .and_then(|state| {
            state
                .selections
                .get(crate::core::protocol::contracts::DECISION_EVENT_FAMILY)
                .map(|selection| selection.version)
        })
        .unwrap_or_else(|| {
            crate::core::protocol::contracts::baseline(
                crate::core::protocol::contracts::DECISION_EVENT_FAMILY,
            )
            .expect("catalogue validates one decision-event baseline")
        });
    if !batch.iter().all(|event| {
        crate::core::protocol::contracts::decision_event_can_write(
            &event.envelope,
            selected_version,
        )
    }) {
        selected_version =
            renegotiate_event_writer(client, config, key, openlatch_dir, egress).await?;
    }
    if !batch.iter().all(|event| {
        crate::core::protocol::contracts::decision_event_can_write(
            &event.envelope,
            selected_version,
        )
    }) {
        record_event_compatibility(
            openlatch_dir,
            None,
            &format!("decision_event v{selected_version} would discard record-v2 evidence"),
        );
        return Err(CloudError::CompatibilityUnavailable);
    }

    // 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,
        config.host_key.as_deref(),
        config.agent_id.as_deref(),
    );

    // 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");

    // The egress outcome is recorded HERE, at the send site, because this is the
    // only place the transport error still exists — `CloudError::Network` erases
    // it. Any response at all, including a 401 or a 500, proves the route works:
    // what the server said about the payload is not this module's business.
    let response = match client
        .post(&url)
        .headers(headers)
        .header(
            crate::core::protocol::contracts::SUPPORT_HEADER,
            crate::core::protocol::contracts::support_header_value(),
        )
        .body(body)
        .send()
        .await
    {
        Ok(response) => {
            egress.record_ok(&url);
            response
        }
        Err(e) => {
            egress.record_failure(&url, &e);
            return Err(CloudError::Network);
        }
    };

    let status = response.status();

    if status.is_success() {
        if let Some(raw) = response
            .headers()
            .get(crate::core::protocol::contracts::SELECTED_HEADER)
        {
            let selected = raw
                .to_str()
                .ok()
                .and_then(|raw| crate::core::protocol::contracts::parse_selected(raw).ok())
                .and_then(|map| {
                    map.get(crate::core::protocol::contracts::DECISION_EVENT_FAMILY)
                        .copied()
                });
            let Some(selected) = selected else {
                record_event_compatibility(
                    openlatch_dir,
                    None,
                    "invalid or incomplete selected-map acknowledgement",
                );
                return Err(CloudError::CompatibilityUnavailable);
            };
            if !crate::core::protocol::contracts::has_writer(
                crate::core::protocol::contracts::DECISION_EVENT_FAMILY,
                selected,
            ) {
                record_event_compatibility(
                    openlatch_dir,
                    None,
                    &format!("selected decision_event v{selected} has no local writer"),
                );
                return Err(CloudError::CompatibilityUnavailable);
            }
            if let Err(error) = crate::core::protocol::contracts::record_selection(
                openlatch_dir,
                crate::core::protocol::contracts::DECISION_EVENT_FAMILY,
                selected,
            ) {
                tracing::debug!(%error, "could not persist decision-event selection");
            }
        }
        return Ok(());
    }

    if status.as_u16() == 409 {
        let value = response.json::<serde_json::Value>().await.ok();
        if value.as_ref().is_some_and(|problem| {
            problem.get("type").and_then(serde_json::Value::as_str)
                == Some(crate::core::protocol::contracts::COMPATIBILITY_PROBLEM_TYPE)
        }) {
            let platform_range = value
                .as_ref()
                .and_then(|problem| problem.get("platform_range"))
                .and_then(event_problem_range);
            record_event_compatibility(
                openlatch_dir,
                platform_range,
                "no common retained representation",
            );
            return Err(CloudError::CompatibilityUnavailable);
        }
        return Err(CloudError::ClientError(409));
    }

    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 })
        }
        // A licence refusal, whatever the body says (I-4). 402 is
        // RFC-undefined, so an unparseable body is still a refusal — retaining
        // the batch is the safe answer, and dropping it (which is what the old
        // `ClientError` arm did) is the one that loses events over a renewal.
        402 => Err(license_refused(&response.headers().clone(), response).await),
        // A 5xx is a server fault UNLESS its body names a site-licence code.
        // The body is read once, bounded by the client's own response limits,
        // and only for these two statuses — no other non-2xx body is read.
        503 => {
            let headers = response.headers().clone();
            let body = response.text().await.unwrap_or_default();
            if body_code_is_site_license(&body) {
                Err(license_refused_from(&headers, &body))
            } else {
                Err(CloudError::ServerError)
            }
        }
        500..=599 => Err(CloudError::ServerError),
        code => Err(CloudError::ClientError(code)),
    }
}

/// Read a refusal response and turn it into [`CloudError::LicenseRefused`].
async fn license_refused(
    headers: &reqwest::header::HeaderMap,
    response: reqwest::Response,
) -> CloudError {
    let body = response.text().await.unwrap_or_default();
    license_refused_from(headers, &body)
}

/// The pure half: the platform's house error envelope plus its `Retry-After`.
///
/// The structured fields ride `error.details[0]` (the platform's own
/// convention), with `error.licensing_url` accepted as a flat fallback so a
/// simpler body is not silently stripped of its remedy.
///
/// An `error.code` that is neither a `license_*` nor a `site_license_*` code
/// nor `clock_regression` becomes `license_unknown` rather than being passed
/// through: the code reaches a doctor remedy and `openlatch status`, and an
/// arbitrary string from an intermediary is not something to render there.
pub(crate) fn license_refused_from(headers: &reqwest::header::HeaderMap, body: &str) -> CloudError {
    let parsed: Option<serde_json::Value> = serde_json::from_str(body).ok();
    let error = parsed.as_ref().and_then(|v| v.get("error"));

    let code = error
        .and_then(|e| e.get("code"))
        .and_then(serde_json::Value::as_str)
        .filter(|code| code.starts_with("license_") || SITE_LICENSE_CODES.contains(code))
        .unwrap_or(LICENSE_UNKNOWN_CODE)
        .to_string();

    let licensing_url = error
        .and_then(|e| {
            e.pointer("/details/0/licensing_url")
                .or_else(|| e.get("licensing_url"))
        })
        .and_then(serde_json::Value::as_str)
        .map(str::to_string);

    let retry_after = headers
        .get(reqwest::header::RETRY_AFTER)
        .and_then(|v| v.to_str().ok())
        .and_then(parse_retry_after)
        .map(|secs| Duration::from_secs(secs.min(MAX_RETRY_AFTER_SECS)));

    CloudError::LicenseRefused {
        code,
        licensing_url,
        retry_after,
    }
}

/// Re-open event-family negotiation without serializing or sending the queued
/// event facts. An empty CloudEvents batch carries only the support header; a
/// concrete acknowledgement is required before the caller may retry its real
/// batch under a different writer.
async fn renegotiate_event_writer(
    client: &reqwest::Client,
    config: &CloudConfig,
    key: &SecretString,
    openlatch_dir: &Path,
    egress: &crate::egress::EgressReporter,
) -> Result<u32, CloudError> {
    let request_id = uuid::Uuid::now_v7().to_string();
    let headers = build_cloud_headers(
        key,
        &request_id,
        config.host_key.as_deref(),
        config.agent_id.as_deref(),
    );
    let url = format!(
        "{}/api/v1/events/ingest",
        config.api_url.trim_end_matches('/')
    );
    let response = match client
        .post(&url)
        .headers(headers)
        .header(
            crate::core::protocol::contracts::SUPPORT_HEADER,
            crate::core::protocol::contracts::support_header_value(),
        )
        .body("[]")
        .send()
        .await
    {
        Ok(response) => {
            egress.record_ok(&url);
            response
        }
        Err(error) => {
            egress.record_failure(&url, &error);
            return Err(CloudError::Network);
        }
    };

    let status = response.status();
    if status.is_success() {
        let selected = response
            .headers()
            .get(crate::core::protocol::contracts::SELECTED_HEADER)
            .and_then(|raw| raw.to_str().ok())
            .and_then(|raw| crate::core::protocol::contracts::parse_selected(raw).ok())
            .and_then(|map| {
                map.get(crate::core::protocol::contracts::DECISION_EVENT_FAMILY)
                    .copied()
            });
        let Some(selected) = selected.filter(|version| {
            crate::core::protocol::contracts::has_writer(
                crate::core::protocol::contracts::DECISION_EVENT_FAMILY,
                *version,
            )
        }) else {
            record_event_compatibility(
                openlatch_dir,
                None,
                "event renegotiation returned no usable decision_event acknowledgement",
            );
            return Err(CloudError::CompatibilityUnavailable);
        };
        if let Err(error) = crate::core::protocol::contracts::record_selection(
            openlatch_dir,
            crate::core::protocol::contracts::DECISION_EVENT_FAMILY,
            selected,
        ) {
            tracing::debug!(%error, "could not persist renegotiated decision-event selection");
        }
        return Ok(selected);
    }

    if status.as_u16() == 409 {
        let value = response.json::<serde_json::Value>().await.ok();
        if value.as_ref().is_some_and(|problem| {
            problem.get("type").and_then(serde_json::Value::as_str)
                == Some(crate::core::protocol::contracts::COMPATIBILITY_PROBLEM_TYPE)
        }) {
            let platform_range = value
                .as_ref()
                .and_then(|problem| problem.get("platform_range"))
                .and_then(event_problem_range);
            record_event_compatibility(
                openlatch_dir,
                platform_range,
                "no common retained representation",
            );
            return Err(CloudError::CompatibilityUnavailable);
        }
        record_event_compatibility(
            openlatch_dir,
            None,
            "event renegotiation returned an unregistered 409 response",
        );
        return Err(CloudError::CompatibilityUnavailable);
    }

    match status.as_u16() {
        401 | 403 => Err(CloudError::AuthError),
        // A licence refusal is not a compatibility failure: surface it as one.
        402 => Err(license_refused(&response.headers().clone(), response).await),
        429 => {
            let retry_after_secs = response
                .headers()
                .get("Retry-After")
                .and_then(|value| value.to_str().ok())
                .and_then(|value| value.parse::<u64>().ok())
                .unwrap_or(config.rate_limit_default_secs);
            Err(CloudError::RateLimit { retry_after_secs })
        }
        500..=599 => Err(CloudError::ServerError),
        code => {
            record_event_compatibility(
                openlatch_dir,
                None,
                &format!("event renegotiation was refused with HTTP {code}"),
            );
            Err(CloudError::CompatibilityUnavailable)
        }
    }
}

fn event_problem_range(
    value: &serde_json::Value,
) -> Option<crate::core::protocol::contracts::VersionRange> {
    use crate::core::protocol::contracts::VersionRange;
    if let Some(raw) = value.as_str() {
        return crate::core::protocol::contracts::parse_ranges(&format!("peer={raw}"))
            .ok()?
            .get("peer")
            .copied();
    }
    if let Some(values) = value.as_array() {
        return VersionRange::new(
            u32::try_from(values.first()?.as_u64()?).ok()?,
            u32::try_from(values.get(1)?.as_u64()?).ok()?,
        );
    }
    VersionRange::new(
        u32::try_from(value.get("oldest")?.as_u64()?).ok()?,
        u32::try_from(value.get("newest")?.as_u64()?).ok()?,
    )
}

fn record_event_compatibility(
    openlatch_dir: &Path,
    platform_range: Option<crate::core::protocol::contracts::VersionRange>,
    detail: &str,
) {
    use crate::core::protocol::contracts::{self, CompatibilityDiagnostic, DECISION_EVENT_FAMILY};
    let Some(client_range) = contracts::supported_ranges()
        .get(DECISION_EVENT_FAMILY)
        .copied()
    else {
        return;
    };
    let diagnostic = CompatibilityDiagnostic {
        family: DECISION_EVENT_FAMILY.to_string(),
        client_range,
        platform_range,
        last_selection: None,
        observed_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
        detail: detail.to_string(),
    };
    if let Err(error) = contracts::record_diagnostic(openlatch_dir, diagnostic) {
        tracing::debug!(%error, "could not persist decision-event compatibility diagnostic");
    }
}

/// 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 crate::core::supervision::task::{
        spawn_supervised, Backoff, HealthRegistry, RestartPolicy, TaskSpec, TaskState,
    };
    use secrecy::SecretString;
    use std::sync::atomic::Ordering;
    use std::sync::Mutex;
    use tokio::sync::mpsc;

    /// An empty session registry — what a worker test that is not about session
    /// stamping passes. No live session means the egress falls back to the
    /// install-id bucket, which is the shape these tests already assert.
    fn no_sessions() -> Arc<SessionRegistry> {
        Arc::new(SessionRegistry::default())
    }

    /// Poll `cond` until it is true or `timeout` elapses.
    ///
    /// A fixed `sleep(N).await` before an assertion bets that the worker's
    /// background task gets N ms of real CPU time to make progress — a bet
    /// that a host sharing cores with other `cargo test` runs loses (issue
    /// #380). This polls for the signal the test actually claims instead:
    /// it returns as soon as `cond` holds, so a healthy run is exactly as
    /// fast as before, and only a genuinely starved run spends more of the
    /// bound — never a wider tolerance on what the assertion accepts.
    async fn wait_until<F: FnMut() -> bool>(mut cond: F, timeout: std::time::Duration) -> bool {
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            if cond() {
                return true;
            }
            if tokio::time::Instant::now() >= deadline {
                return false;
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
    }

    // ---------------------------------------------------------------------------
    // 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,
        /// Number of `invalidate()` calls — the worker's 401/403 latch must
        /// call this exactly once per latch onset (issue #306, contract 2).
        invalidations: 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),
                invalidations: std::sync::atomic::AtomicU64::new(0),
            })
        }

        fn empty() -> Arc<Self> {
            Arc::new(Self {
                key: Mutex::new(None),
                retrievals: std::sync::atomic::AtomicU64::new(0),
                invalidations: 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)
        }

        fn invalidations(&self) -> u64 {
            self.invalidations.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())))
        }

        fn invalidate(&self) {
            self.invalidations.fetch_add(1, Ordering::Relaxed);
        }
    }

    #[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, &crate::egress::EgressConfig::direct()).expect("client");
        // The client is valid if no panic occurred — reqwest doesn't expose internals for assertion
    }

    /// The worker used to `.expect()` here, so an egress config it could not honour took
    /// the daemon down. It now reports, and `run_cloud_worker_on` stops forwarding.
    #[test]
    fn build_cloud_client_reports_a_refused_route_instead_of_panicking() {
        let mut egress = crate::egress::EgressConfig::direct();
        egress.mode = crate::egress::ProxyMode::Auto;
        egress.allow_direct = false;
        let err = build_cloud_client(&CloudConfig::default(), &egress)
            .expect_err("no proxy plus allow_direct = false must refuse");
        assert_eq!(err.code, crate::error::ERR_DIRECT_FORBIDDEN);
    }

    #[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(),
            crate::egress::EgressReporter::direct(),
            state,
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // 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(),
            crate::egress::EgressReporter::direct(),
            state,
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // 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());
    }

    // -----------------------------------------------------------------------
    // The 402 contract (I-4)
    //
    // The invariant under test throughout: a licence refusal is never a
    // credential failure and never stops the daemon. Events are RETAINED, the
    // gate holds POSTs on a cadence rather than a 30-second ladder, and the
    // first POST that succeeds afterwards clears it.
    // -----------------------------------------------------------------------

    const LICENSING_URL: &str = "https://app.openlatch.ai/settings/licensing";

    /// The platform's house error envelope, exactly as I-2's gate sends it.
    const I2_LICENSE_BODY: &str = concat!(
        r#"{"error":{"code":"license_expired","message":"This organization's licence has expired.","#,
        r#""details":[{"licensing_url":"https://app.openlatch.ai/settings/licensing"}]}}"#
    );

    /// The one config every licence case runs on, shared by the worker and the
    /// drain task so both talk to the same mock server on the same cadence.
    /// `batch_max_wait_ms` is small so a single event flushes on the time
    /// trigger without the test having to close the channel first.
    fn license_config(server: &mockito::Server) -> CloudConfig {
        CloudConfig {
            api_url: server.url(),
            batch_max_wait_ms: 50,
            retry_delay_ms: 5,
            ..Default::default()
        }
    }

    /// A worker wired to `server`, with an outbox on a tempdir.
    ///
    /// Returns the pieces every case below asserts on.
    fn license_worker(
        server: &mockito::Server,
        dir: &tempfile::TempDir,
        outbox: Option<Arc<Outbox>>,
    ) -> (
        mpsc::Sender<CloudEvent>,
        CloudState,
        tokio::task::JoinHandle<()>,
    ) {
        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let state = CloudState::new();
        let config = license_config(server);
        let handle = tokio::spawn(run_cloud_worker(
            rx,
            TestCredentialProvider::with_key("test-api-key"),
            config,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            outbox,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));
        (tx, state, handle)
    }

    /// The durable-outbox drain, spawned beside [`license_worker`]. It is its
    /// own supervised subsystem (issue #306, contract 3) rather than a loop
    /// nested in `run_cloud_worker`, so a test that exercises the *replay*
    /// path has to spawn it alongside, exactly as `daemon::mod` does.
    ///
    /// Split from `license_worker` rather than folded into it: the tests that
    /// only exercise the live POST path assert exact request counts, and a
    /// drain task replaying the same outbox would add requests they never
    /// asked for. Spawning it separately also lets a test arm the licence gate
    /// *before* the drain's immediate startup pass runs.
    ///
    /// The returned `watch::Sender` must outlive the test body — dropping it
    /// resolves the loop's shutdown arm and ends the drain.
    fn spawn_license_drain(
        server: &mockito::Server,
        dir: &tempfile::TempDir,
        outbox: Arc<Outbox>,
        state: CloudState,
    ) -> (
        tokio::task::JoinHandle<()>,
        tokio::sync::watch::Sender<bool>,
    ) {
        let config = license_config(server);
        let drain_client = crate::egress::ClientHandle::of(
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap(),
        );
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let handle = tokio::spawn(run_outbox_drain_on(
            outbox,
            drain_client,
            config,
            TestCredentialProvider::with_key("test-api-key"),
            state,
            dir.path().to_path_buf(),
            SourceFormats::new(),
            no_sessions(),
            crate::egress::EgressReporter::direct(),
            shutdown_rx,
            None,
        ));
        (handle, shutdown_tx)
    }

    async fn send_event(tx: &mpsc::Sender<CloudEvent>, id: &str) {
        let _ = tx
            .send(CloudEvent {
                envelope: serde_json::json!({ "id": id }),
                agent_id: "agt_test".to_string(),
            })
            .await;
    }

    /// The headline case. A 402 must spool rather than drop, must not latch
    /// `auth_error` in memory **or on disk**, and must not touch the drop
    /// counter — which is exactly what the old `ClientError(402)` path did.
    #[tokio::test]
    async fn a_402_retains_the_batch_and_is_never_an_auth_error() {
        use std::sync::atomic::Ordering;

        let mut server = mockito::Server::new_async().await;
        let _mock = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(402)
            .with_header("Retry-After", "900")
            .with_body(I2_LICENSE_BODY)
            .create_async()
            .await;

        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));
        let (tx, state, handle) = license_worker(&server, &dir, Some(outbox.clone()));

        send_event(&tx, "evt_licensed").await;
        tokio::time::sleep(Duration::from_millis(250)).await;

        assert_eq!(outbox.pending_count(), 1, "the event must be retained");
        assert_eq!(
            state.drop_count(),
            0,
            "a retained event is not a drop — the outbox has it"
        );
        assert!(
            !state.auth_error.load(Ordering::Relaxed),
            "a licence refusal is never a credential failure"
        );
        // The worker writes `cloud_state.json` on its own schedule, so the
        // assertion is on its CONTENT: whatever else happened, a licence
        // refusal must never have latched an auth error into it.
        let state_path = dir.path().join("cloud_state.json");
        if state_path.exists() {
            let persisted: serde_json::Value =
                serde_json::from_str(&std::fs::read_to_string(&state_path).unwrap()).unwrap();
            assert_eq!(
                persisted["auth_error"], false,
                "a licence refusal must not persist an auth error"
            );
        }

        let gate = state.license_gate().expect("the gate must be set");
        assert_eq!(gate.code, "license_expired");
        assert_eq!(gate.licensing_url.as_deref(), Some(LICENSING_URL));
        let wait = gate
            .until
            .saturating_duration_since(std::time::Instant::now());
        assert!(
            wait > Duration::from_secs(880) && wait <= Duration::from_secs(900),
            "the gate must honour Retry-After: 900, got {wait:?}"
        );

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

    /// While the gate holds, the worker does not POST at all — `expect(1)`
    /// against three events is the assertion. Without the pre-group guard this
    /// would be one request per flush for the life of an expired contract.
    #[tokio::test]
    async fn a_standing_gate_stops_the_worker_posting_at_all() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(402)
            .with_header("Retry-After", "900")
            .with_body(I2_LICENSE_BODY)
            .expect(1)
            .create_async()
            .await;

        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));
        let (tx, state, handle) = license_worker(&server, &dir, Some(outbox.clone()));

        for id in ["evt_1", "evt_2", "evt_3"] {
            send_event(&tx, id).await;
            tokio::time::sleep(Duration::from_millis(120)).await;
        }

        assert!(state.license_gate_active());
        assert_eq!(
            outbox.pending_count(),
            3,
            "every event is queued, not just the one that met the refusal"
        );
        mock.assert_async().await;

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

    /// With no outbox the events really are gone, so they ARE drops — both the
    /// batch that met the refusal and every batch the gate holds afterwards.
    /// This is the one place `record_drops` fires on this path.
    #[tokio::test]
    async fn with_no_outbox_a_refusal_counts_drops() {
        let mut server = mockito::Server::new_async().await;
        let _mock = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(402)
            .with_header("Retry-After", "900")
            .with_body(I2_LICENSE_BODY)
            .create_async()
            .await;

        let dir = tempfile::tempdir().unwrap();
        let (tx, state, handle) = license_worker(&server, &dir, None);

        send_event(&tx, "evt_a").await;
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(state.drop_count(), 1);

        send_event(&tx, "evt_b").await;
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(state.drop_count(), 2, "a gated flush drops too");

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

    /// A refusal met on a RETRY is still a refusal: gated and spooled, never
    /// dropped. Without the arm on each retry ladder this batch would fall into
    /// the shared `Err(_)` and be counted as a drop.
    #[tokio::test]
    async fn a_500_then_a_402_on_the_retry_is_gated_not_dropped() {
        let mut server = mockito::Server::new_async().await;
        let _first = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(500)
            .with_body("boom")
            .expect(1)
            .create_async()
            .await;
        let _second = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(402)
            .with_body(I2_LICENSE_BODY)
            .create_async()
            .await;

        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));
        let (tx, state, handle) = license_worker(&server, &dir, Some(outbox.clone()));

        send_event(&tx, "evt_retry").await;
        tokio::time::sleep(Duration::from_millis(400)).await;

        assert!(state.license_gate().is_some(), "the retry's refusal gates");
        assert_eq!(outbox.pending_count(), 1);
        assert_eq!(state.drop_count(), 0);

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

    /// The drain is the other path that can meet a refusal — and on an idle
    /// host it is the ONLY one. It must set the gate, keep the entry, and
    /// charge no attempt against `OUTBOX_MAX_ATTEMPTS`; charging would
    /// quarantine a whole backlog over a renewal.
    ///
    /// Driven through the real drain task rather than by calling
    /// `drain_outbox_once`, because the gate skip lives in that task's notify
    /// loop.
    #[tokio::test]
    async fn a_drain_that_meets_a_refusal_keeps_its_entries_and_charges_nothing() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(402)
            .with_header("Retry-After", "900")
            .with_body(I2_LICENSE_BODY)
            .expect(1)
            .create_async()
            .await;

        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));
        outbox
            .append(&serde_json::json!({"id": "evt_queued"}))
            .unwrap();

        let (tx, state, handle) = license_worker(&server, &dir, Some(outbox.clone()));
        let (drain_handle, _drain_shutdown) =
            spawn_license_drain(&server, &dir, outbox.clone(), state.clone());

        // The startup drain meets the 402 and sets the gate; every notify after
        // it must be skipped rather than replayed into the refusal.
        for _ in 0..5 {
            state.notify_drain();
            tokio::time::sleep(Duration::from_millis(80)).await;
        }

        mock.assert_async().await;
        assert!(state.license_gate_active());
        assert_eq!(
            outbox.pending_count(),
            1,
            "the entry is retained, never quarantined"
        );

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

    /// A successful drain is what clears the gate on an idle host. Without it
    /// `/metrics` would keep reporting `license_blocked`, `doctor` would stay
    /// Degraded and `status` would keep exiting 7 until some live hook event
    /// happened to be forwarded.
    #[tokio::test]
    async fn a_successful_drain_clears_the_gate() {
        let mut server = mockito::Server::new_async().await;
        let _health = server
            .mock("GET", "/api/v1/health")
            .with_status(200)
            .with_body("{}")
            .create_async()
            .await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(200)
            .with_body("{}")
            .expect_at_least(1)
            .create_async()
            .await;

        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));
        outbox
            .append(&serde_json::json!({"id": "evt_backlog"}))
            .unwrap();

        let (tx, state, handle) = license_worker(&server, &dir, Some(outbox.clone()));
        // A gate whose cadence has already elapsed: a POST is due, and the gate
        // is still reported until one succeeds. Armed before the drain task is
        // spawned, so the gate is in place for its immediate startup pass.
        state.set_license_gate("license_expired".to_string(), None, Some(Duration::ZERO));
        assert!(state.license_gate().is_some());
        let (drain_handle, _drain_shutdown) =
            spawn_license_drain(&server, &dir, outbox.clone(), state.clone());

        for _ in 0..10 {
            if outbox.pending_count() == 0 && state.license_gate().is_none() {
                break;
            }
            state.notify_drain();
            tokio::time::sleep(Duration::from_millis(80)).await;
        }

        mock.assert_async().await;
        assert_eq!(outbox.pending_count(), 0, "the backlog replayed");
        assert!(
            state.license_gate().is_none(),
            "a successful POST is what proves the refusal is over"
        );

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

    #[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 provider_handle = provider.clone();
        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,
            crate::egress::EgressReporter::direct(),
            state_clone,
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // 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;

        // Poll for the 401 latch instead of guessing how long the worker
        // needs (issue #380).
        assert!(
            wait_until(
                || state.auth_error.load(Ordering::Relaxed),
                std::time::Duration::from_secs(5),
            )
            .await,
            "auth_error must be true after 401 response"
        );

        // The latch onset must invalidate the credential provider's cache
        // exactly once (issue #306, contract 2) — otherwise the process-wide
        // keyring memo keeps handing back the same rejected key forever, and
        // a real rotation sitting in the keychain never becomes visible.
        assert_eq!(
            provider_handle.invalidations(),
            1,
            "the 401 latch must invalidate the credential provider exactly once"
        );

        // 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(),
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // No events sent: the correction must come from starting up, not from
        // a successful POST. Poll the file instead of guessing how long
        // startup needs under load (issue #380).
        let cleared = wait_until(
            || {
                std::fs::read_to_string(&state_path)
                    .ok()
                    .and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok())
                    .is_some_and(|v| v["auth_error"] == false)
            },
            std::time::Duration::from_secs(5),
        )
        .await;
        assert!(
            cleared,
            "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,
            crate::egress::EgressReporter::direct(),
            state,
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        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,
            crate::egress::EgressReporter::direct(),
            state,
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        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,
            crate::egress::EgressReporter::direct(),
            state_clone,
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // Health tick fires every 50ms in test builds — poll for it instead
        // of guessing how long one tick needs under load (issue #380).
        assert!(
            wait_until(
                || state.consecutive_drops() == 0,
                std::time::Duration::from_secs(5),
            )
            .await,
            "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,
            crate::egress::EgressReporter::direct(),
            state_clone,
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // Wait for `run_health_probe` to have actually recorded the failure,
        // not just for mockito to have observed the GET — the request
        // landing at the mock says nothing about whether the client-side
        // code finished processing the 5xx response yet, and the assertion
        // below cares about that completed state (issue #380 review).
        assert!(
            wait_until(
                || state.consecutive_probe_failures() >= 1,
                std::time::Duration::from_secs(5),
            )
            .await,
            "at least one health probe failure must be recorded"
        );

        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;
    }

    /// **The load-point proof for the worker.** The daemon builds this client once, so
    /// without the per-flush re-read a healed route would never reach the highest-volume
    /// egress consumer on the host. The handle starts empty (`[proxy] allow_direct = false`
    /// with nothing working), the batch is spooled instead of sent, and the client
    /// installed afterwards — what a self-heal pass installs — carries the next one.
    #[tokio::test]
    async fn a_client_swapped_into_the_handle_is_used_by_the_next_flush() {
        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"}"#)
            // At least one: the spooled event is replayed by the drain as soon as a route
            // exists, so the healed run legitimately produces a second POST.
            .expect_at_least(1)
            .create_async()
            .await;

        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 1_000_000));
        let clients = crate::egress::EgressClients::new(
            crate::egress::Timeouts::default(),
            crate::egress::Timeouts::default(),
        );
        let config = CloudConfig {
            api_url: server.url(),
            batch_max_wait_ms: 50,
            ..Default::default()
        };

        let (tx, rx) = mpsc::channel::<CloudEvent>(10);
        let cloud_state = CloudState::new();
        let worker = {
            let mut rx = rx;
            let config = config.clone();
            let cloud_state = cloud_state.clone();
            let dir = dir.path().to_path_buf();
            let outbox = outbox.clone();
            let handle = clients.cloud.clone();
            tokio::spawn(async move {
                run_cloud_worker_on(
                    &mut rx,
                    TestCredentialProvider::with_key("test-api-key"),
                    config,
                    crate::egress::EgressReporter::direct(),
                    cloud_state,
                    dir,
                    Some(outbox),
                    None,
                    handle,
                    SourceFormats::new(),
                    no_sessions(),
                )
                .await
            })
        };

        let send = |n: &str| CloudEvent {
            envelope: serde_json::json!({ "id": n }),
            agent_id: "agt_test".to_string(),
        };

        let _ = tx.send(send("evt_no_route")).await;
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
        assert_eq!(
            cloud_state.forwarded_count(),
            0,
            "with no permitted route the batch must be spooled, never sent"
        );

        clients
            .apply(&crate::egress::EgressConfig::direct())
            .expect("install a route");

        let _ = tx.send(send("evt_healed")).await;
        for _ in 0..40 {
            if cloud_state.forwarded_count() > 0 {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert!(
            cloud_state.forwarded_count() >= 1,
            "the next flush must use the newly installed client"
        );

        drop(tx);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), worker).await;
        mock_ingest.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,
            crate::egress::EgressReporter::direct(),
            state_clone,
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

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

        // Poll for the forward instead of guessing how long it needs under
        // load (issue #380).
        assert!(
            wait_until(
                || state.forwarded_count() == 1,
                std::time::Duration::from_secs(5),
            )
            .await,
            "event must be forwarded"
        );
        assert_eq!(
            state.consecutive_drops(),
            0,
            "successful forward must clear consecutive_drops"
        );
        assert_eq!(state.drop_count(), 3, "lifetime drop_count preserved");

        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;

        const CHANNEL_CAPACITY: usize = 10;
        let (tx, rx) = mpsc::channel::<CloudEvent>(CHANNEL_CAPACITY);
        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,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // 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;
        }
        // Wait until the worker has actually CONSUMED all three, rather than sleeping a
        // fixed 30ms and hoping it did. This is the difference between the two ways
        // forwarded_count() can read 0: the events were skipped (what this test means), or
        // they are simply still sitting in the channel (what it must not accept). In the
        // second case the credential arrives first, the worker then finds a key waiting,
        // and all three get forwarded -- the server sees four POSTs. That is exactly how
        // this passed locally and failed on a loaded CI runner.
        let consumed = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            while tx.capacity() < CHANNEL_CAPACITY {
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await;
        assert!(
            consumed.is_ok(),
            "worker never drained the three credential-less events ({} slots still queued)",
            CHANNEL_CAPACITY - tx.capacity()
        );
        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 drain_provider = provider.clone();
        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.clone(),
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            Some(outbox.clone()),
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // The outbox drain is its own supervised subsystem (issue #306,
        // contract 3), not nested inside the worker above — spawn it
        // alongside, as `daemon::mod` does.
        let drain_client = crate::egress::ClientHandle::of(
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap(),
        );
        let (_drain_shutdown_tx, drain_shutdown_rx) = tokio::sync::watch::channel(false);
        let drain_handle = tokio::spawn(run_outbox_drain_on(
            outbox.clone(),
            drain_client,
            config,
            drain_provider,
            state.clone(),
            dir.path().to_path_buf(),
            SourceFormats::new(),
            no_sessions(),
            crate::egress::EgressReporter::direct(),
            drain_shutdown_rx,
            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;
        drain_handle.abort();

        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 drain_provider = provider.clone();
        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.clone(),
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            Some(outbox.clone()),
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // The outbox drain is its own supervised subsystem (issue #306,
        // contract 3), woken by the worker's own health-probe `notify_drain()`
        // above but no longer run inside it — spawn it alongside, as
        // `daemon::mod` does.
        let drain_client = crate::egress::ClientHandle::of(
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap(),
        );
        let (_drain_shutdown_tx, drain_shutdown_rx) = tokio::sync::watch::channel(false);
        let drain_handle = tokio::spawn(run_outbox_drain_on(
            outbox.clone(),
            drain_client,
            config,
            drain_provider,
            state.clone(),
            dir.path().to_path_buf(),
            SourceFormats::new(),
            no_sessions(),
            crate::egress::EgressReporter::direct(),
            drain_shutdown_rx,
            None,
        ));

        // Live POST fails twice (initial + retry) → envelope lands in outbox.
        // Poll instead of guessing how long the retry ladder needs under
        // load (issue #380).
        let _ = tx
            .send(CloudEvent {
                envelope: serde_json::json!({"id": "evt_offline_1"}),
                agent_id: "agt_test".to_string(),
            })
            .await;
        assert!(
            wait_until(
                || outbox.pending_count() == 1,
                std::time::Duration::from_secs(5),
            )
            .await,
            "envelope must be spooled after the failed retries"
        );
        mock_fail_ingest.assert_async().await;
        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. Poll for the drain instead of
        // guessing how long it needs after the recovery signal fires under
        // load (issue #380).
        assert!(
            wait_until(|| mock_replay.matched(), std::time::Duration::from_secs(5)).await,
            "the drain must replay the spooled event after recovery"
        );

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

        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()
    }

    /// Every envelope `id` across all captured batches, order-independent —
    /// for asserting exactly *which* events landed, not just how many.
    fn captured_ids(captured: &Captured) -> std::collections::BTreeSet<String> {
        captured
            .lock()
            .unwrap()
            .iter()
            .flat_map(|body| {
                serde_json::from_str::<Vec<serde_json::Value>>(body)
                    .expect("every request body must be a JSON array")
                    .into_iter()
                    .map(|v| v["id"].as_str().unwrap_or_default().to_string())
                    .collect::<Vec<_>>()
            })
            .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,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        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,
            crate::egress::EgressReporter::direct(),
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        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 = 1300`: push at t=0, push at
    /// t=1000, observe at t=1800. Anchored to the first event the flush is
    /// due at t=1300 and has happened — 500ms of slack for the real HTTP
    /// round trip (mockito async server) to complete on a loaded CI runner.
    /// Reset on every push it would be due at t=2300 and nothing would have
    /// been sent — which is exactly how `batch_max_wait_ms` silently stops
    /// bounding latency under load; observing 500ms before that reset
    /// deadline keeps the two cases unambiguous.
    #[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: 1300,
            ..Default::default()
        };

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

        tx.send(evt("evt_first")).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
        tx.send(evt("evt_second")).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(800)).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,
            crate::egress::EgressReporter::direct(),
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        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,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        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();
        // Poll for the time-triggered flush instead of guessing how long it
        // needs under load (issue #380).
        assert!(
            wait_until(
                || state.forwarded_count() == 2,
                std::time::Duration::from_secs(5),
            )
            .await,
            "the two normal events must be forwarded"
        );

        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,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            Some(outbox.clone()),
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        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 provider = TestCredentialProvider::with_key("test-api-key");
        let drain_provider = provider.clone();
        let state = CloudState::new();

        let handle = tokio::spawn(run_cloud_worker(
            rx,
            provider,
            config.clone(),
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            Some(outbox.clone()),
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // The outbox drain is its own supervised subsystem (issue #306,
        // contract 3), not nested inside the worker above — spawn it
        // alongside, as `daemon::mod` does. It fires an immediate pass at
        // its own startup; no live events are sent, so every request below
        // comes from the replay path.
        let drain_client = crate::egress::ClientHandle::of(
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap(),
        );
        let (_drain_shutdown_tx, drain_shutdown_rx) = tokio::sync::watch::channel(false);
        let drain_handle = tokio::spawn(run_outbox_drain_on(
            outbox.clone(),
            drain_client,
            config,
            drain_provider,
            state,
            dir.path().to_path_buf(),
            SourceFormats::new(),
            no_sessions(),
            crate::egress::EgressReporter::direct(),
            drain_shutdown_rx,
            None,
        ));
        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;
        drain_handle.abort();
    }

    /// issue #306, contract 3 — reproduction + fix proof.
    ///
    /// Before the fix, the outbox drain was a bare `tokio::spawn` nested
    /// inside `run_cloud_worker_on` (worker.rs, pre-fix ~line 403): a panic
    /// in it was invisible (nothing joined the handle, the cloud worker's own
    /// supervised task kept reporting `Running`) and permanent (the function
    /// spawns the drain exactly once, at its own startup). This test injects
    /// one panic into [`run_outbox_drain_on`], now its own supervised
    /// subsystem, and asserts both halves of the acceptance criterion: the
    /// panic is *observed* on the health registry (a restart is recorded),
    /// and the drain *recovers* — the very next `notify_drain()` actually
    /// replays a real entry.
    ///
    /// The failpoint (`fail_once`) is an `Arc<AtomicBool>` owned by this test
    /// alone, never a process-global static — a parallel test's own drain
    /// task cannot consume it or be crashed by it (issue #306 review).
    #[tokio::test]
    async fn outbox_drain_panic_is_observed_and_recovers() {
        use crate::core::cloud::outbox::Outbox;

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

        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));
        let cloud_state = CloudState::new();
        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };
        let egress = crate::egress::EgressReporter::direct();
        let client = crate::egress::ClientHandle::of(
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap(),
        );
        let provider: Arc<dyn CredentialProvider> =
            TestCredentialProvider::with_key("test-api-key");

        let registry = Arc::new(HealthRegistry::new());
        let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);

        // Arm exactly one panic: the very first drain pass this task runs
        // (its own immediate startup pass) blows up. Instance-owned, so it
        // cannot race a neighbouring test's own drain task.
        let fail_once = Arc::new(std::sync::atomic::AtomicBool::new(true));

        let outbox_for_task = outbox.clone();
        let cloud_state_for_task = cloud_state.clone();
        let task_shutdown_rx = shutdown_rx.clone();
        spawn_supervised(
            &registry,
            TaskSpec::new("outbox-drain-test", RestartPolicy::Always).with_backoff(Backoff::new(
                Duration::from_millis(5),
                Duration::from_millis(20),
            )),
            shutdown_rx,
            move || {
                run_outbox_drain_on(
                    outbox_for_task.clone(),
                    client.clone(),
                    config.clone(),
                    provider.clone(),
                    cloud_state_for_task.clone(),
                    dir.path().to_path_buf(),
                    SourceFormats::new(),
                    no_sessions(),
                    egress.clone(),
                    task_shutdown_rx.clone(),
                    Some(fail_once.clone()),
                )
            },
        );

        // Give the panic time to land and the supervisor time to restart.
        tokio::time::sleep(Duration::from_millis(200)).await;
        let health = registry.tasks()[0].clone();
        assert!(
            health.restarts() >= 1,
            "a panicking outbox-drain pass must be recorded as a restart on the \
             health registry — on main this task had no health entry at all, got {}",
            health.restarts()
        );

        // Prove it actually recovered, not just that the supervisor spun: a
        // fresh entry plus the next notify_drain() must be drained for real.
        outbox
            .append(&serde_json::json!({"id": "evt_after_recovery"}))
            .unwrap();
        cloud_state.notify_drain();
        tokio::time::sleep(Duration::from_millis(300)).await;

        assert_eq!(
            outbox.pending_count(),
            0,
            "the outbox must drain after the drain task recovers from its panic"
        );
        assert!(
            !batch_sizes(&captured).is_empty(),
            "a replay POST must go out after recovery"
        );
    }

    /// issue #306 review, contract 3 — the drain task must end cleanly on
    /// daemon shutdown rather than parking forever on `notify.notified()`
    /// until the supervisor's `SHUTDOWN_DRAIN` window (4s) elapses and force-
    /// aborts it. This asserts the join lands well under that window, which
    /// only holds if the task's own `select!` observed the shutdown signal.
    #[tokio::test]
    async fn outbox_drain_stops_cleanly_on_shutdown() {
        use crate::core::cloud::outbox::Outbox;

        let dir = tempfile::tempdir().unwrap();
        let outbox = Arc::new(Outbox::new(dir.path(), 0));
        let cloud_state = CloudState::new();
        let config = CloudConfig::default();
        let egress = crate::egress::EgressReporter::direct();
        let client = crate::egress::ClientHandle::of(
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap(),
        );
        let provider: Arc<dyn CredentialProvider> =
            TestCredentialProvider::with_key("test-api-key");

        let registry = Arc::new(HealthRegistry::new());
        let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
        let task_shutdown_rx = shutdown_rx.clone();

        let handle = spawn_supervised(
            &registry,
            TaskSpec::new("outbox-drain-shutdown-test", RestartPolicy::Always).with_backoff(
                Backoff::new(Duration::from_millis(5), Duration::from_millis(20)),
            ),
            shutdown_rx,
            move || {
                run_outbox_drain_on(
                    outbox.clone(),
                    client.clone(),
                    config.clone(),
                    provider.clone(),
                    cloud_state.clone(),
                    dir.path().to_path_buf(),
                    SourceFormats::new(),
                    no_sessions(),
                    egress.clone(),
                    task_shutdown_rx.clone(),
                    None,
                )
            },
        );

        // Let the immediate startup pass finish, then signal shutdown.
        tokio::time::sleep(Duration::from_millis(100)).await;
        shutdown_tx.send(true).expect("shutdown send");

        // Comfortably under the supervisor's 4s force-abort window — this
        // only passes if the task's own select! observed the signal and
        // returned, rather than the supervisor timing out and aborting it.
        let joined = tokio::time::timeout(Duration::from_millis(500), handle).await;
        assert!(
            joined.is_ok(),
            "the drain task must stop within 500ms of shutdown, not be force-aborted \
             after the supervisor's drain window"
        );

        assert_eq!(registry.tasks()[0].state(), TaskState::Stopped);
    }

    /// issue #306 re-review, contract 3 — `biased;` must make shutdown win
    /// whenever it is ready at the same time as an already-stored notify
    /// permit (e.g. a health-recovery `notify_drain()` landing right as
    /// `openlatch stop` fires). Without `biased;`, `select!`'s randomized
    /// poll order lets the notify arm win on some fraction of runs, starting
    /// one more drain pass after shutdown was already signalled.
    ///
    /// `TestCredentialProvider::retrievals()` is the counter: `drain_outbox_once`
    /// fetches the credential unconditionally on every pass, before it even
    /// looks at the outbox, so it counts passes regardless of whether there
    /// was anything to drain. Exactly one pass — the unconditional startup
    /// pass — is correct; a second means the loop ran again after shutdown.
    ///
    /// Both signals are armed only after a short sleep lets the supervisor
    /// spawn the task and the task park inside its `select!`, so this is a
    /// genuine "both ready when polled" race, not a fixed order of events —
    /// repeated across several trials since a race's absence is not provable
    /// from one instance of it going the right way.
    #[tokio::test]
    async fn outbox_drain_shutdown_wins_a_race_against_notify() {
        use crate::core::cloud::outbox::Outbox;

        for trial in 0..20 {
            let dir = tempfile::tempdir().unwrap();
            let outbox = Arc::new(Outbox::new(dir.path(), 0));
            let cloud_state = CloudState::new();
            let config = CloudConfig::default();
            let egress = crate::egress::EgressReporter::direct();
            let client = crate::egress::ClientHandle::of(
                build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap(),
            );
            let provider = TestCredentialProvider::with_key("test-api-key");
            let provider_for_task = provider.clone();
            let cloud_state_for_task = cloud_state.clone();

            let registry = Arc::new(HealthRegistry::new());
            let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
            let task_shutdown_rx = shutdown_rx.clone();

            let handle = spawn_supervised(
                &registry,
                TaskSpec::new("outbox-drain-race-test", RestartPolicy::Always).with_backoff(
                    Backoff::new(Duration::from_millis(5), Duration::from_millis(20)),
                ),
                shutdown_rx,
                move || {
                    run_outbox_drain_on(
                        outbox.clone(),
                        client.clone(),
                        config.clone(),
                        provider_for_task.clone(),
                        cloud_state_for_task.clone(),
                        dir.path().to_path_buf(),
                        SourceFormats::new(),
                        no_sessions(),
                        egress.clone(),
                        task_shutdown_rx.clone(),
                        None,
                    )
                },
            );

            // Let the startup pass finish and the loop park inside select!.
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Fire both before yielding again: a stored notify permit and an
            // already-true shutdown flag are then both ready the instant the
            // parked select! is next polled.
            cloud_state.notify_drain();
            shutdown_tx.send(true).expect("shutdown send");

            let joined = tokio::time::timeout(Duration::from_millis(500), handle).await;
            assert!(
                joined.is_ok(),
                "trial {trial}: drain task must still stop promptly under the race"
            );

            assert_eq!(
                provider.retrievals(),
                1,
                "trial {trial}: shutdown racing an already-stored notify must not start \
                 an extra drain pass — only the unconditional startup pass should have run"
            );
        }
    }

    /// 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,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            Some(shutdown_rx),
            SourceFormats::new(),
            no_sessions(),
        ));

        for i in 0..7 {
            tx.send(evt(&format!("evt_{i}"))).await.unwrap();
        }
        // Every `.send().await` above already completed by the time this
        // line runs — the event is enqueued the moment the call returns, so
        // there is nothing further to wait for here. Do NOT wait for the
        // worker to have *received* them too: that would hide the exact
        // race this test exists to catch. The worker's shutdown branch is
        // `biased`-listed ahead of `rx.recv()`, so signalling shutdown
        // immediately after these sends return is what used to abandon
        // events still sitting, sent-but-unread, in the channel — the
        // shutdown branch now drains them itself before flushing (issue
        // #380 review).
        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);
    }

    /// C5, continued — the shutdown drain is bounded, not just non-blocking.
    /// A sender that keeps refilling the channel after shutdown fires must
    /// not stop the worker from exiting, and none of the refill may be
    /// counted: the drain snapshots `rx.len()` once and takes exactly that
    /// many (issue #380 review).
    ///
    /// This needs genuine OS-level concurrency, not just a second tokio
    /// task: the drain loop itself has no `.await` inside it, so on a
    /// `current_thread` runtime it always runs as one uninterrupted burst —
    /// nothing else, bounded or not, can be scheduled mid-loop, and the old
    /// unbounded `while let Ok(..) = rx.try_recv()` would pass just as
    /// cleanly as the fix. A `std::thread` spinning `try_send` on a
    /// `multi_thread` runtime can actually race the drain on a different
    /// core, so this is the version that tells the two implementations
    /// apart (confirmed by hand: swapping the fix back to the old unbounded
    /// loop makes this test hang past the bound).
    ///
    /// The channel's capacity is set to exactly the pre-shutdown burst, so
    /// it is completely full the instant shutdown is signalled — every
    /// refill attempt can therefore only land once the drain frees a slot,
    /// and FIFO ordering means a slot the drain frees is filled by whatever
    /// the *bounded* loop has not yet reached, never by something the drain
    /// then also consumes: the snapshot fixes the loop at exactly
    /// `PRE_SHUTDOWN` iterations regardless of what the racing thread
    /// manages to inject into freed slots afterward.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn shutdown_drain_is_bounded_against_a_refilling_sender() {
        let mut server = mockito::Server::new_async().await;
        let _health = quiet_health(&mut server).await;
        let (_ingest, captured) = capture_ingest(&mut server, 200).await;

        const PRE_SHUTDOWN: usize = 7;
        let (tx, rx) = mpsc::channel::<CloudEvent>(PRE_SHUTDOWN);
        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,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            Some(shutdown_rx),
            SourceFormats::new(),
            no_sessions(),
        ));

        for i in 0..PRE_SHUTDOWN {
            tx.send(evt(&format!("evt_{i}"))).await.unwrap();
        }

        // A real OS thread, not a tokio task, spinning `try_send` from
        // before shutdown fires until the channel closes. It can genuinely
        // run in parallel with the drain on the second worker thread —
        // that parallelism is the point.
        let refill_tx = tx.clone();
        let refill = std::thread::spawn(move || {
            let mut sent = 0u64;
            let mut i = 0u64;
            loop {
                match refill_tx.try_send(evt(&format!("post_{i}"))) {
                    Ok(()) => {
                        sent += 1;
                        i += 1;
                    }
                    Err(mpsc::error::TrySendError::Full(_)) => {
                        std::thread::yield_now();
                    }
                    Err(mpsc::error::TrySendError::Closed(_)) => break,
                }
            }
            sent
        });

        // The channel is completely full the moment the refill thread
        // starts (it was already full from the sends above), so shutdown
        // races a producer that has been trying continuously, not one that
        // starts afterward.
        shutdown_tx.send(true).unwrap();

        let exited = tokio::time::timeout(std::time::Duration::from_secs(5), handle).await;
        assert!(
            exited.is_ok(),
            "worker must exit within the bound despite a refilling sender"
        );

        // Dropping the last sender closes the channel, which is what stops
        // the refill thread's spin loop (`Closed`).
        drop(tx);
        let refill_sent = refill.join().expect("refill thread must not panic");

        assert_eq!(
            captured_ids(&captured),
            (0..PRE_SHUTDOWN)
                .map(|i| format!("evt_{i}"))
                .collect::<std::collections::BTreeSet<_>>(),
            "the forwarded ids must be exactly the pre-shutdown burst, none of the {refill_sent} refill sends"
        );
        assert_eq!(state.forwarded_count(), PRE_SHUTDOWN as u64);
    }

    /// 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,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // 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,
            crate::egress::EgressReporter::direct(),
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        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:
    /// `model_relay/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 `model_relay::emit::assemble_event` builds — no
        // `clientversion` anywhere in it.
        let event = CloudEvent {
            envelope: serde_json::json!({
                "specversion": "1.0",
                "id": "evt_model_relay",
                "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,
            &SourceFormats::new(),
            &SessionRegistry::default(),
            None,
        );

        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");
    }

    /// D-02's totality gate. `wireformat` is an attribute of the AGENT, so it
    /// rides a **hook** envelope — not just an economics one — and it is stamped
    /// here, at the single egress, rather than in each builder. Per-builder
    /// stamping is exactly how `clientversion` became accidentally partial.
    ///
    /// The map is supplied as a LITERAL by the test. It has to be: a developer
    /// machine's `detect_agents()` finds no Codex, so a global built from the
    /// filesystem would make this test unwritable.
    #[test]
    fn wireformat_is_stamped_on_a_hook_event() {
        let formats: SourceFormats = [
            ("claude-code", WireFormat::AnthropicMessages),
            ("codex-cli", WireFormat::OpenAiResponses),
        ]
        .into_iter()
        .collect();

        let event = CloudEvent {
            envelope: serde_json::json!({
                "specversion": "1.0",
                "id": "evt_hook",
                "source": "codex-cli",
                "type": "ai.openlatch.hook.pre_tool_use",
                "data": {"tool_name": "Bash"},
            }),
            agent_id: "agt_test".to_string(),
        };

        let stamped = stamp_extensions(&event, &formats, &SessionRegistry::default(), None);

        assert_eq!(
            stamped["wireformat"].as_str(),
            Some("openai-responses"),
            "the agent's format must ride a HOOK event, not only an economics one"
        );
        // The neighbours are untouched, and nothing the producer set moved.
        assert_eq!(stamped["agentid"].as_str(), Some("agt_test"));
        assert_eq!(stamped["data"]["tool_name"], "Bash");
    }

    /// **The one a green suite hides.** After a provider switch, an envelope
    /// must carry the NEW format with no daemon restart.
    ///
    /// `SourceFormats` used to be a map moved into the worker, so it learned the
    /// answer once at daemon start. That was harmless while every binding
    /// returned a constant, and became a live defect the moment a GUI-hosted
    /// agent's format followed whichever provider the customer is on: they
    /// switch in the UI, the wiring supervisor re-wires them on its next tick,
    /// and every hook, config and economics envelope from then until a restart
    /// is labelled with the protocol they left.
    ///
    /// The handle is shared, so the update the daemon makes at the wiring site
    /// is visible to the worker's next batch. This test holds the two ends of
    /// that share.
    /// **A producer that watched the request wins over the per-source map.**
    ///
    /// The map answers per SOURCE, so it can only describe an agent wired to one
    /// request plane. Cline carries many providers at once and therefore has no
    /// entry at all: its economics rows were stamped `unknown` while their own
    /// `data` named the provider the request actually went to — `google`, on the
    /// row this test is shaped after.
    #[test]
    fn a_producer_that_named_the_format_keeps_it() {
        // No entry for `cline`, which is the real state of this map for an agent
        // wired per provider slot.
        let formats = SourceFormats::new();
        let relay_row = CloudEvent {
            envelope: serde_json::json!({
                "specversion": "1.0",
                "id": "evt_econ",
                "source": "cline",
                "type": "ai.openlatch.economics.usage",
                "wireformat": "google-generate-content",
                "data": {"gen_ai.provider.name": "google"},
            }),
            agent_id: "agt_test".to_string(),
        };
        assert_eq!(
            stamp_extensions(&relay_row, &formats, &SessionRegistry::default(), None)["wireformat"]
                .as_str(),
            Some("google-generate-content"),
            "the listener served the request and knows which protocol it spoke"
        );

        // A hook event from the same agent names no format, and the map has no
        // answer either — so it still reports `unknown` rather than borrowing
        // the relay's.
        let hook_event = CloudEvent {
            envelope: serde_json::json!({
                "specversion": "1.0",
                "id": "evt_hook",
                "source": "cline",
                "type": "pre_tool_use",
                "data": {"toolName": "execute_command"},
            }),
            agent_id: "agt_test".to_string(),
        };
        assert_eq!(
            stamp_extensions(&hook_event, &formats, &SessionRegistry::default(), None)
                ["wireformat"]
                .as_str(),
            Some("unknown"),
            "always present, and never a value nobody observed"
        );

        // An empty string is not a producer naming anything.
        let blank = CloudEvent {
            envelope: serde_json::json!({
                "specversion": "1.0", "id": "evt_blank", "source": "claude-code",
                "type": "pre_tool_use", "wireformat": "",
            }),
            agent_id: "agt_test".to_string(),
        };
        let formats: SourceFormats = [("claude-code", WireFormat::AnthropicMessages)]
            .into_iter()
            .collect();
        assert_eq!(
            stamp_extensions(&blank, &formats, &SessionRegistry::default(), None)["wireformat"]
                .as_str(),
            Some("anthropic-messages")
        );
    }

    #[test]
    fn source_formats_invalidated_on_provider_change() {
        let formats: SourceFormats = [("cline", WireFormat::OpenAiChatCompletions)]
            .into_iter()
            .collect();
        // The worker's own copy — cloned exactly as the daemon clones it.
        let worker_view = formats.clone();

        let event = CloudEvent {
            envelope: serde_json::json!({
                "specversion": "1.0",
                "id": "evt_hook",
                "source": "cline",
                "type": "ai.openlatch.hook.pre_tool_use",
                "data": {"tool_name": "Bash"},
            }),
            agent_id: "agt_test".to_string(),
        };

        assert_eq!(
            stamp_extensions(&event, &worker_view, &SessionRegistry::default(), None)["wireformat"]
                .as_str(),
            Some("openai-chat-completions")
        );

        // The customer switches provider; the wiring supervisor re-wires and
        // records the new format at the one site that knows it changed.
        formats.set("cline", WireFormat::GoogleGenerateContent);

        assert_eq!(
            stamp_extensions(&event, &worker_view, &SessionRegistry::default(), None)["wireformat"]
                .as_str(),
            Some("google-generate-content"),
            "the worker's view must follow, with no restart — a stale label here is \
             mis-attributed economics for every turn until the daemon is bounced"
        );
    }

    /// ALWAYS PRESENT (owner decision, 2026-09-05). A source we cannot resolve
    /// carries the literal `"unknown"` rather than omitting the key, so the
    /// platform can group and join on the column with no null handling — the
    /// same shape `ai.openlatch.session.source` already uses on the very same
    /// row. This is also every unattributed Model relay event, since
    /// `resolved_source()` answers `"unknown"` there.
    #[test]
    fn wireformat_is_unknown_for_an_unknown_source() {
        let formats: SourceFormats = [("claude-code", WireFormat::AnthropicMessages)]
            .into_iter()
            .collect();

        for envelope in [
            // The model relay's unattributed economics row.
            serde_json::json!({"id": "evt_a", "source": "unknown"}),
            // An agent with no binding on this host.
            serde_json::json!({"id": "evt_b", "source": "codex-cli"}),
            // No `source` at all.
            serde_json::json!({"id": "evt_c"}),
        ] {
            let stamped = stamp_extensions(
                &CloudEvent {
                    envelope: envelope.clone(),
                    agent_id: "agt_test".to_string(),
                },
                &formats,
                &SessionRegistry::default(),
                None,
            );
            // The KEY must exist, not merely compare equal — an assertion on the
            // value alone passes against a missing key in some accessor styles,
            // which is the exact failure this test replaced.
            assert!(
                stamped.get("wireformat").is_some(),
                "the key is never omitted: {envelope}"
            );
            assert_eq!(
                stamped["wireformat"].as_str(),
                Some("unknown"),
                "{envelope}"
            );
        }
    }

    /// 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,
            &SourceFormats::new(),
            &SessionRegistry::default(),
            None,
        );

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

    // ---------------------------------------------------------------------------
    // The `hostid` fill at egress (I-4)
    // ---------------------------------------------------------------------------

    /// A 32-hex value shaped exactly like a real `hostid`.
    const HOST: &str = "9262b296baa37a205734509345581251";

    /// The config monitor, the tamper reconciler and the model relay never pass
    /// through `daemon/handlers.rs`, so without this fill the platform keys
    /// their envelopes `agent:<agent_id>` and counts a second licensed host
    /// beside the machine's real one.
    #[test]
    fn a_producerless_envelope_leaves_the_egress_carrying_the_host_id() {
        let stamped = stamp_extensions(
            &config_event(),
            &SourceFormats::new(),
            &SessionRegistry::default(),
            Some(HOST),
        );
        assert_eq!(stamped["hostid"].as_str(), Some(HOST));
    }

    /// `or_insert_with`, not `insert`: a hook envelope keeps exactly what
    /// `host_stamp` put there. Overwriting would make the egress the authority
    /// on an attribute the hook path already owns.
    #[test]
    fn an_envelope_that_already_carries_a_host_id_keeps_it() {
        const OTHER: &str = "deadbeefdeadbeefdeadbeefdeadbeef";
        let event = CloudEvent {
            envelope: serde_json::json!({"id": "evt_hook", "hostid": OTHER}),
            agent_id: "agt_install".to_string(),
        };

        let stamped = stamp_extensions(
            &event,
            &SourceFormats::new(),
            &SessionRegistry::default(),
            Some(HOST),
        );
        assert_eq!(stamped["hostid"].as_str(), Some(OTHER));
    }

    /// A host with no machine identifier gets no `hostid` on either path —
    /// absent, never empty and never the host key's `agent:` fallback.
    #[test]
    fn no_host_id_configured_fills_nothing() {
        let stamped = stamp_extensions(
            &config_event(),
            &SourceFormats::new(),
            &SessionRegistry::default(),
            None,
        );
        assert!(stamped.get("hostid").is_none());
    }

    // ---------------------------------------------------------------------------
    // Session stamping (`stamp_session`)
    // ---------------------------------------------------------------------------

    /// A config-monitor-shaped envelope: no `subject`, `data` we author.
    fn config_event() -> CloudEvent {
        CloudEvent {
            envelope: serde_json::json!({
                "id": "evt_cfg",
                "source": "claude-code",
                "type": "ai.openlatch.config.modified",
                "data": {"configkind": "settings"},
            }),
            agent_id: "agt_install".to_string(),
        }
    }

    fn assurance_of(stamped: &serde_json::Value) -> Option<&str> {
        stamped["data"][crate::model_relay::session::ASSURANCE_KEY].as_str()
    }

    #[test]
    fn one_live_session_is_stamped_attested_on_a_subjectless_event() {
        let sessions = SessionRegistry::default();
        sessions.upsert("agt_install", "agt_install", "claude-code", "sess_only");

        let stamped = stamp_extensions(&config_event(), &SourceFormats::new(), &sessions, None);

        assert_eq!(
            stamped["subject"].as_str(),
            Some("sess_only"),
            "a config event must land in the session that was live when it happened"
        );
        assert_eq!(assurance_of(&stamped), Some("attested"));
    }

    /// Two live sessions and the registry has to pick, so the row says so. The
    /// correlation is still worth making — the alternative is an event the
    /// platform cannot display at all — but it must never read as a fact.
    #[test]
    fn concurrent_sessions_are_stamped_inferred() {
        let sessions = SessionRegistry::default();
        sessions.upsert("agt_install", "agt_install", "claude-code", "sess_old");
        std::thread::sleep(std::time::Duration::from_millis(5));
        sessions.upsert("agt_install", "agt_install", "claude-code", "sess_new");

        let stamped = stamp_extensions(&config_event(), &SourceFormats::new(), &sessions, None);

        assert_eq!(stamped["subject"].as_str(), Some("sess_new"));
        assert_eq!(assurance_of(&stamped), Some("inferred"));
    }

    /// A session belonging to ANOTHER agent is not a candidate.
    ///
    /// The reported defect: on a host running Claude Code, Codex and Cline at
    /// once, Claude Code's config and tamper events were stamped with whichever
    /// session had been active most recently — a Cline conversation id — and
    /// the platform, whose sessions are agent-scoped, rendered one session
    /// holding three agents. `agent_id` is the INSTALL, shared by every agent
    /// on the host, so resolving by it alone could never tell them apart.
    #[test]
    fn a_session_of_a_different_agent_is_never_borrowed() {
        let sessions = SessionRegistry::default();
        sessions.upsert("agt_install", "agt_install", "cline", "conv_cline_1");

        let stamped = stamp_extensions(&config_event(), &SourceFormats::new(), &sessions, None);

        assert_eq!(
            stamped["subject"].as_str(),
            Some("agt_install"),
            "a Claude Code event falls back to the install bucket rather than \
             joining a live Cline conversation"
        );
        assert_eq!(
            assurance_of(&stamped),
            Some("unknown"),
            "and says plainly that no session of its own agent was claimed"
        );
    }

    /// The scope narrows the candidates; it does not stop attribution.
    #[test]
    fn the_events_own_agent_still_wins_its_session() {
        let sessions = SessionRegistry::default();
        sessions.upsert("agt_install", "agt_install", "cline", "conv_cline_1");
        std::thread::sleep(std::time::Duration::from_millis(5));
        sessions.upsert("agt_install", "agt_install", "claude-code", "sess_cc");
        std::thread::sleep(std::time::Duration::from_millis(5));
        // Cline is the most recently active, and must still lose.
        sessions.upsert("agt_install", "agt_install", "cline", "conv_cline_2");

        let stamped = stamp_extensions(&config_event(), &SourceFormats::new(), &sessions, None);

        assert_eq!(stamped["subject"].as_str(), Some("sess_cc"));
        assert_eq!(assurance_of(&stamped), Some("attested"));
    }

    /// No agent was running — someone edited the file by hand, or this is one of
    /// the ~38 `config.snapshot` events `init` writes before any agent exists.
    /// The install id is a stable per-host bucket, deliberately preferred to a
    /// null that would hide the event; `unknown` says no session was claimed.
    #[test]
    fn no_live_session_buckets_the_event_under_the_install_id() {
        let event = config_event();

        let stamped = stamp_extensions(
            &event,
            &SourceFormats::new(),
            &SessionRegistry::default(),
            None,
        );

        assert_eq!(stamped["subject"].as_str(), Some("agt_install"));
        assert_eq!(stamped["subject"].as_str(), Some(event.agent_id.as_str()));
        assert_eq!(assurance_of(&stamped), Some("unknown"));
    }

    /// `config.removed` builds its envelope with `data: Value::Null`. The
    /// annotation still has to land somewhere, so the null becomes the object it
    /// would have been.
    #[test]
    fn a_null_data_becomes_the_object_the_assurance_lands_in() {
        let event = CloudEvent {
            envelope: serde_json::json!({
                "id": "evt_removed",
                "type": "ai.openlatch.config.removed",
                "data": serde_json::Value::Null,
            }),
            agent_id: "agt_install".to_string(),
        };

        let stamped = stamp_extensions(
            &event,
            &SourceFormats::new(),
            &SessionRegistry::default(),
            None,
        );

        assert_eq!(assurance_of(&stamped), Some("unknown"));
    }

    /// A hook envelope names its own session, and its `data` is the agent's raw
    /// payload — opaque, never rewritten (AGENTS.md). So the producer's subject
    /// stands and nothing is added beside it: the platform reads an absent
    /// assurance on a hook row as attested by construction, since the agent
    /// handed us the id itself.
    #[test]
    fn a_hook_events_subject_and_raw_payload_are_left_alone() {
        let sessions = SessionRegistry::default();
        sessions.upsert("agt_install", "agt_install", "claude-code", "sess_other");
        let data = serde_json::json!({"session_id": "sess_hook", "tool_name": "Bash"});
        let event = CloudEvent {
            envelope: serde_json::json!({
                "id": "evt_hook",
                "source": "claude-code",
                "type": "ai.openlatch.hook.pre_tool_use",
                "subject": "sess_hook",
                "data": data.clone(),
            }),
            agent_id: "agt_install".to_string(),
        };

        let stamped = stamp_extensions(&event, &SourceFormats::new(), &sessions, None);

        assert_eq!(
            stamped["subject"].as_str(),
            Some("sess_hook"),
            "the agent's own session id must not be replaced by a correlation"
        );
        assert_eq!(stamped["data"], data, "the agent's raw payload is opaque");
    }

    /// The model relay resolves its own session through the same registry, and
    /// records how sure the cascade was. Re-resolving here would relabel an
    /// honest `inferred` as `attested` whenever the concurrency happened to have
    /// died down by flush time.
    #[test]
    fn a_model_relay_row_keeps_the_assurance_the_cascade_recorded() {
        let sessions = SessionRegistry::default();
        sessions.upsert("agt_install", "agt_install", "claude-code", "sess_live");
        let event = CloudEvent {
            envelope: serde_json::json!({
                "id": "evt_usage",
                "source": "claude-code",
                "type": "ai.openlatch.economics.usage",
                "subject": "sess_model_relay",
                "data": {
                    crate::model_relay::session::ASSURANCE_KEY: "inferred",
                },
            }),
            agent_id: "agt_install".to_string(),
        };

        let stamped = stamp_extensions(&event, &SourceFormats::new(), &sessions, None);

        assert_eq!(stamped["subject"].as_str(), Some("sess_model_relay"));
        assert_eq!(assurance_of(&stamped), Some("inferred"));
    }

    /// Replay must not change a spooled envelope's session. Entries are stamped
    /// before they are spooled, so the drain's re-stamp sees a subject already
    /// present and leaves it — even when the session it names died overnight.
    #[test]
    fn re_stamping_a_spooled_envelope_is_idempotent() {
        let sessions = SessionRegistry::default();
        sessions.upsert("agt_install", "agt_install", "claude-code", "sess_first");
        let once = stamp_extensions(&config_event(), &SourceFormats::new(), &sessions, None);

        // The overnight outage: the original session is gone, another is live.
        let later = SessionRegistry::default();
        later.upsert("agt_install", "agt_install", "claude-code", "sess_second");
        let replayed = stamp_extensions(
            &CloudEvent {
                envelope: once.clone(),
                agent_id: "agt_install".to_string(),
            },
            &SourceFormats::new(),
            &later,
            None,
        );

        assert_eq!(replayed["subject"], once["subject"]);
        assert_eq!(assurance_of(&replayed), Some("attested"));
    }

    /// 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,
            crate::egress::EgressReporter::direct(),
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // Mixed producers: unversioned (model_relay/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");
    }

    /// Which machine is asking, on the ingest route. The mock matches the
    /// header rather than asserting on the response, so a header sent under the
    /// wrong name fails here instead of silently reaching the platform.
    #[tokio::test]
    async fn ingest_carries_the_host_key_and_omits_it_when_there_is_none() {
        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()];

        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .match_header("x-openlatch-machine-id", HOST)
            .with_status(200)
            .with_body("{}")
            .expect(1)
            .create_async()
            .await;
        let config = CloudConfig {
            api_url: server.url(),
            host_key: Some(HOST.to_string()),
            ..Default::default()
        };
        let client =
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).expect("client");
        post_batch(
            &client,
            &config,
            &key,
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .expect("a matched request answers 200");
        mock.assert_async().await;

        // No host key means no header at all — the platform reads its presence
        // as "this host identified itself".
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .match_header("x-openlatch-machine-id", mockito::Matcher::Missing)
            .with_status(200)
            .with_body("{}")
            .expect(1)
            .create_async()
            .await;
        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };
        let client =
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).expect("client");
        post_batch(
            &client,
            &config,
            &key,
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .expect("a request with no machine id answers 200");
        mock.assert_async().await;
    }

    /// The install id, on both ingest requests: the real batch and the `[]`
    /// renegotiation probe. The probe carries no envelope at all, so without
    /// this header its acknowledgement lands attributed to no install
    /// (platform#915) until the next real batch replays.
    #[tokio::test]
    async fn ingest_carries_the_agent_id_on_the_batch_and_the_probe_and_omits_it_when_there_is_none(
    ) {
        const AGENT: &str = "agt_test_448";
        let key = SecretString::from("test-api-key".to_string());
        let batch = vec![prepare_one(serde_json::json!({"id": "evt_1"})).unwrap()];

        // The real batch POST.
        let dir = tempfile::tempdir().unwrap();
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .match_header("x-openlatch-agent-id", AGENT)
            .with_status(200)
            .with_body("{}")
            .expect(1)
            .create_async()
            .await;
        let config = CloudConfig {
            api_url: server.url(),
            agent_id: Some(AGENT.to_string()),
            ..Default::default()
        };
        let client =
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).expect("client");
        post_batch(
            &client,
            &config,
            &key,
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .expect("a matched request answers 200");
        mock.assert_async().await;

        // The `[]` renegotiation probe.
        let dir = tempfile::tempdir().unwrap();
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .match_body("[]")
            .match_header("x-openlatch-agent-id", AGENT)
            .with_status(200)
            .with_header(
                crate::core::protocol::contracts::SELECTED_HEADER,
                "decision_event=1",
            )
            .with_body("{}")
            .expect(1)
            .create_async()
            .await;
        let config = CloudConfig {
            api_url: server.url(),
            agent_id: Some(AGENT.to_string()),
            ..Default::default()
        };
        let client =
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).expect("client");
        renegotiate_event_writer(
            &client,
            &config,
            &key,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .expect("a matched probe answers 200");
        mock.assert_async().await;

        // No agent id (e.g. before `init`) means no header at all — never
        // sent empty.
        let dir = tempfile::tempdir().unwrap();
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/api/v1/events/ingest")
            .match_header("x-openlatch-agent-id", mockito::Matcher::Missing)
            .with_status(200)
            .with_body("{}")
            .expect(1)
            .create_async()
            .await;
        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };
        let client =
            build_cloud_client(&config, &crate::egress::EgressConfig::direct()).expect("client");
        post_batch(
            &client,
            &config,
            &key,
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .expect("a request with no agent id answers 200");
        mock.assert_async().await;
    }

    /// 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, body, expected) in [
            (401u16, None, "{}", CloudError::AuthError),
            (403, None, "{}", CloudError::AuthError),
            (
                429,
                Some(("Retry-After", "7")),
                "{}",
                CloudError::RateLimit {
                    retry_after_secs: 7,
                },
            ),
            // -- licence refusals (I-4) --------------------------------------
            //
            // The platform's house envelope: the structured fields ride
            // `details[0]`, and the `Retry-After` is the gate's own cadence.
            (
                402,
                Some(("Retry-After", "900")),
                I2_LICENSE_BODY,
                CloudError::LicenseRefused {
                    code: "license_expired".to_string(),
                    licensing_url: Some(LICENSING_URL.to_string()),
                    retry_after: Some(Duration::from_secs(900)),
                },
            ),
            // A flatter body still yields its remedy — the fallback exists so a
            // simpler shape is not silently stripped of the URL.
            (
                402,
                None,
                r#"{"error":{"code":"license_expired","licensing_url":"https://app.openlatch.ai/settings/licensing"}}"#,
                CloudError::LicenseRefused {
                    code: "license_expired".to_string(),
                    licensing_url: Some(LICENSING_URL.to_string()),
                    retry_after: None,
                },
            ),
            // 402 is RFC-undefined, so an unparseable body is STILL a refusal —
            // retaining the batch is the safe reading (D-09).
            (
                402,
                None,
                "<html>402</html>",
                CloudError::LicenseRefused {
                    code: "license_unknown".to_string(),
                    licensing_url: None,
                    retry_after: None,
                },
            ),
            // A `Retry-After` past the cap is clamped, never honoured.
            (
                402,
                Some(("Retry-After", "99999")),
                I2_LICENSE_BODY,
                CloudError::LicenseRefused {
                    code: "license_expired".to_string(),
                    licensing_url: Some(LICENSING_URL.to_string()),
                    retry_after: Some(Duration::from_secs(MAX_RETRY_AFTER_SECS)),
                },
            ),
            (
                503,
                None,
                r#"{"error":{"code":"site_license_expired"}}"#,
                CloudError::LicenseRefused {
                    code: "site_license_expired".to_string(),
                    licensing_url: None,
                    retry_after: None,
                },
            ),
            (
                503,
                None,
                r#"{"error":{"code":"clock_regression"}}"#,
                CloudError::LicenseRefused {
                    code: "clock_regression".to_string(),
                    licensing_url: None,
                    retry_after: None,
                },
            ),
            // A 503 with no site code is still an ordinary server fault.
            (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(body).create_async().await;

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

        // An HTTP-date `Retry-After` is legal (RFC 9110 §10.2.3) and the gate
        // may send one. Its own case because the value has to be computed now.
        {
            let mut server = mockito::Server::new_async().await;
            let future = chrono::Utc::now() + chrono::Duration::seconds(600);
            let _mock = server
                .mock("POST", "/api/v1/events/ingest")
                .with_status(402)
                .with_header(
                    "Retry-After",
                    &future.format("%a, %d %b %Y %H:%M:%S GMT").to_string(),
                )
                .with_body(I2_LICENSE_BODY)
                .create_async()
                .await;
            let config = CloudConfig {
                api_url: server.url(),
                ..Default::default()
            };
            let client = build_cloud_client(&config, &crate::egress::EgressConfig::direct())
                .expect("client");
            let err = post_batch(
                &client,
                &config,
                &key,
                &batch,
                dir.path(),
                &crate::egress::EgressReporter::direct(),
            )
            .await
            .expect_err("a 402 must map to a refusal");
            let CloudError::LicenseRefused { retry_after, .. } = err else {
                panic!("an http-date Retry-After must still be a refusal: {err:?}");
            };
            let secs = retry_after.expect("the date form must parse").as_secs();
            assert!(
                (590..=600).contains(&secs),
                "expected ~600s from an http-date, got {secs}"
            );
        }

        // 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, &crate::egress::EgressConfig::direct()).expect("client");
        let err = post_batch(
            &client,
            &config,
            &key,
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .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, &crate::egress::EgressConfig::direct()).expect("client");
        let err = post_batch(
            &client,
            &config,
            &key,
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .unwrap_err();
        assert!(
            matches!(err, CloudError::Network),
            "transport failure must map to Network, got {err:?}"
        );
    }

    #[tokio::test]
    async fn event_negotiation_is_independent_and_no_ack_keeps_the_existing_writer() {
        let mut server = mockito::Server::new_async().await;
        let post = server
            .mock("POST", "/api/v1/events/ingest")
            .match_header(
                crate::core::protocol::contracts::SUPPORT_HEADER,
                "decision_event=1-2,policy_bundle=1-2",
            )
            .with_status(200)
            .expect(1)
            .create_async()
            .await;
        let dir = tempfile::tempdir().unwrap();
        crate::core::protocol::contracts::record_diagnostic(
            dir.path(),
            crate::core::protocol::contracts::CompatibilityDiagnostic {
                family: crate::core::protocol::contracts::POLICY_BUNDLE_FAMILY.to_string(),
                client_range: crate::core::protocol::contracts::VersionRange::new(1, 2).unwrap(),
                platform_range: Some(
                    crate::core::protocol::contracts::VersionRange::new(3, 4).unwrap(),
                ),
                last_selection: None,
                observed_at: chrono::Utc::now().to_rfc3339(),
                detail: "bundle unavailable".to_string(),
            },
        )
        .unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };
        let client = build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap();
        let batch = vec![prepare_one(serde_json::json!({
            "id":"evt-v2", "olverdict":"block", "olresult":"blocked"
        }))
        .unwrap()];
        post_batch(
            &client,
            &config,
            &SecretString::from("key".to_string()),
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .unwrap();
        post.assert_async().await;
        let state = crate::core::protocol::contracts::read_compatibility(dir.path()).unwrap();
        assert!(state
            .diagnostics
            .contains_key(crate::core::protocol::contracts::POLICY_BUNDLE_FAMILY));
        assert!(!state
            .selections
            .contains_key(crate::core::protocol::contracts::DECISION_EVENT_FAMILY));
    }

    #[tokio::test]
    async fn event_ack_drives_next_writer_and_refuses_lossy_lowering() {
        let mut server = mockito::Server::new_async().await;
        let ack = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(200)
            .with_header(
                crate::core::protocol::contracts::SELECTED_HEADER,
                "decision_event=1",
            )
            .expect(1)
            .create_async()
            .await;
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };
        let client = build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap();
        let key = SecretString::from("key".to_string());
        let v1 = vec![prepare_one(serde_json::json!({"id":"evt-v1","olverdict":"allow"})).unwrap()];
        post_batch(
            &client,
            &config,
            &key,
            &v1,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .unwrap();
        ack.assert_async().await;
        assert_eq!(
            crate::core::protocol::contracts::read_compatibility(dir.path())
                .unwrap()
                .selections[crate::core::protocol::contracts::DECISION_EVENT_FAMILY]
                .version,
            1
        );

        let still_v1 = server
            .mock("POST", "/api/v1/events/ingest")
            .match_body("[]")
            .with_status(200)
            .with_header(
                crate::core::protocol::contracts::SELECTED_HEADER,
                "decision_event=1",
            )
            .expect(1)
            .create_async()
            .await;
        let v2 = vec![prepare_one(serde_json::json!({
            "id":"evt-lossy", "olverdict":"ask", "olenforced":0, "olresult":"deferred"
        }))
        .unwrap()];
        let error = post_batch(
            &client,
            &config,
            &key,
            &v2,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .unwrap_err();
        assert!(matches!(error, CloudError::CompatibilityUnavailable));
        still_v1.assert_async().await;
        assert!(
            crate::core::protocol::contracts::read_compatibility(dir.path())
                .unwrap()
                .diagnostics
                .contains_key(crate::core::protocol::contracts::DECISION_EVENT_FAMILY)
        );
    }

    #[tokio::test]
    async fn a_v1_selection_can_renegotiate_to_v2_before_sending_v2_facts() {
        let mut server = mockito::Server::new_async().await;
        let probe = server
            .mock("POST", "/api/v1/events/ingest")
            .match_body("[]")
            .with_status(200)
            .with_header(
                crate::core::protocol::contracts::SELECTED_HEADER,
                "decision_event=2",
            )
            .expect(1)
            .create_async()
            .await;
        let facts = server
            .mock("POST", "/api/v1/events/ingest")
            .match_body(mockito::Matcher::Regex("evt-recovered".to_string()))
            .with_status(200)
            .expect(1)
            .create_async()
            .await;
        let dir = tempfile::tempdir().unwrap();
        crate::core::protocol::contracts::record_selection(
            dir.path(),
            crate::core::protocol::contracts::DECISION_EVENT_FAMILY,
            1,
        )
        .unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };
        let client = build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap();
        let batch = vec![prepare_one(serde_json::json!({
            "id":"evt-recovered", "olverdict":"block", "olenforced":1, "olresult":"blocked"
        }))
        .unwrap()];
        post_batch(
            &client,
            &config,
            &SecretString::from("key".to_string()),
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .unwrap();
        probe.assert_async().await;
        facts.assert_async().await;
        let state = crate::core::protocol::contracts::read_compatibility(dir.path()).unwrap();
        assert_eq!(
            state.selections[crate::core::protocol::contracts::DECISION_EVENT_FAMILY].version,
            2
        );
        assert!(!state
            .diagnostics
            .contains_key(crate::core::protocol::contracts::DECISION_EVENT_FAMILY));
    }

    #[tokio::test]
    async fn a_platform_that_refuses_the_empty_probe_never_receives_v2_facts() {
        let mut server = mockito::Server::new_async().await;
        let probe = server
            .mock("POST", "/api/v1/events/ingest")
            .match_body("[]")
            .with_status(422)
            .expect(1)
            .create_async()
            .await;
        let dir = tempfile::tempdir().unwrap();
        crate::core::protocol::contracts::record_selection(
            dir.path(),
            crate::core::protocol::contracts::DECISION_EVENT_FAMILY,
            1,
        )
        .unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };
        let client = build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap();
        let batch = vec![prepare_one(serde_json::json!({
            "id":"evt-retained", "olverdict":"optimize", "olresult":"rewritten"
        }))
        .unwrap()];
        let error = post_batch(
            &client,
            &config,
            &SecretString::from("key".to_string()),
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .unwrap_err();
        assert!(matches!(error, CloudError::CompatibilityUnavailable));
        probe.assert_async().await;
    }

    #[tokio::test]
    async fn typed_event_mismatch_is_retryable_compatibility_state() {
        let mut server = mockito::Server::new_async().await;
        let mismatch = server
            .mock("POST", "/api/v1/events/ingest")
            .with_status(409)
            .with_body(
                serde_json::json!({
                    "type": crate::core::protocol::contracts::COMPATIBILITY_PROBLEM_TYPE,
                    "family": crate::core::protocol::contracts::DECISION_EVENT_FAMILY,
                    "platform_range": "3-4"
                })
                .to_string(),
            )
            .create_async()
            .await;
        let dir = tempfile::tempdir().unwrap();
        let config = CloudConfig {
            api_url: server.url(),
            ..Default::default()
        };
        let client = build_cloud_client(&config, &crate::egress::EgressConfig::direct()).unwrap();
        let batch = vec![prepare_one(serde_json::json!({"id":"evt-compat"})).unwrap()];
        let error = post_batch(
            &client,
            &config,
            &SecretString::from("key".to_string()),
            &batch,
            dir.path(),
            &crate::egress::EgressReporter::direct(),
        )
        .await
        .unwrap_err();
        assert!(matches!(error, CloudError::CompatibilityUnavailable));
        mismatch.assert_async().await;
        let diagnostic = &crate::core::protocol::contracts::read_compatibility(dir.path())
            .unwrap()
            .diagnostics[crate::core::protocol::contracts::DECISION_EVENT_FAMILY];
        assert_eq!(
            diagnostic.platform_range,
            crate::core::protocol::contracts::VersionRange::new(3, 4)
        );
    }

    /// The invariant behind the idle probe: a host that is actually sending
    /// traffic never issues one, because the traffic already answered the same
    /// question. Asserted on the gate itself rather than by counting requests
    /// over a five-minute window — the arithmetic is the thing that can break,
    /// and it can break in a millisecond.
    #[test]
    fn a_route_with_recent_traffic_issues_no_probe() {
        let cfg = crate::egress::EgressConfig::direct();
        let state = crate::egress::EgressState::new(&cfg);
        let egress = crate::egress::EgressReporter::recording(&cfg, state.clone());

        assert!(
            health_probe_due(&egress),
            "a host that has recorded nothing yet is silent, and silence is what the probe is for"
        );

        state.record_ok();
        assert!(
            !health_probe_due(&egress),
            "traffic is flowing — probing would ask a question already answered"
        );

        // A stream of failures is not silence either: the consumers are already
        // reporting, and a probe would only add load to a link that is down.
        let failing = crate::egress::EgressState::new(&cfg);
        let failing_egress = crate::egress::EgressReporter::recording(&cfg, failing.clone());
        failing.record_failure(crate::error::ERR_PROXY_UNREACHABLE, "refused");
        assert!(!health_probe_due(&failing_egress));
    }

    #[test]
    fn a_route_that_reports_nowhere_probes_exactly_as_before() {
        // Every test in this file, and every non-daemon caller, takes this path.
        assert!(health_probe_due(&crate::egress::EgressReporter::direct()));
    }

    /// 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,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        for i in 0..3 {
            tx.send(evt(&format!("evt_{i}"))).await.unwrap();
        }
        // Poll for the health tick and the credential-poll retrievals instead
        // of guessing how long they need under load (issue #380). Neither
        // trigger can flush the batch regardless of how long this waits
        // (batch_max_wait_ms is out of reach and the size trigger needs 50
        // events), so this cannot race the "must not have flushed" check
        // below.
        assert!(
            wait_until(
                || observer.retrievals() >= 2 && health.matched(),
                std::time::Duration::from_secs(5),
            )
            .await,
            "the health and credential-poll branches must keep firing while a batch waits"
        );

        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,
            crate::egress::EgressReporter::direct(),
            CloudState::new(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        // 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,
            crate::egress::EgressReporter::direct(),
            state.clone(),
            dir.path().to_path_buf(),
            None,
            None,
            SourceFormats::new(),
            no_sessions(),
        ));

        for i in 0..50 {
            tx.send(evt(&format!("evt_{i:02}"))).await.unwrap();
        }
        // Poll for the size-triggered flush instead of guessing how long it
        // needs under load (issue #380).
        assert!(
            wait_until(
                || state.forwarded_count() == 50,
                std::time::Duration::from_secs(5),
            )
            .await,
            "all 50 events must be forwarded"
        );

        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());
    }
}