openlatch-client 0.1.18

//! Cloud event forwarding — types, configuration, and module declarations.
//!
//! This is a strict leaf module. It MUST NOT import from `daemon/`, `cli/`, or
//! any other `core/` sibling module. Only external crates and std are allowed.
//!
//! The cloud forwarding pipeline:
//! 1. Daemon handler calls `try_send(CloudEvent)` into an mpsc channel (non-blocking)
//! 2. `worker::run_cloud_worker` consumes events and POSTs to the cloud API
//! 3. Auth state is persisted to `~/.openlatch/cloud_state.json` for cross-process visibility

pub mod envelope;
pub(crate) mod offset;
pub mod outbox;
pub mod tamper;
pub mod worker;

use std::sync::{
    atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering},
    Arc,
};
use std::time::Duration;

use tokio::sync::Notify;

use secrecy::SecretString;

/// Unix epoch milliseconds — never panics, returns 0 if the clock is
/// before 1970 (impossible on any real system).
#[inline]
pub(crate) fn now_unix_ms() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

/// Unix epoch seconds — same contract as `now_unix_ms`.
#[inline]
pub(crate) fn now_unix_secs() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

// ---------------------------------------------------------------------------
// Credential provider abstraction
// ---------------------------------------------------------------------------

/// Minimal credential provider trait for the cloud worker.
///
/// The worker accepts `Arc<dyn CredentialProvider>` to obtain the API key.
/// In production, a `KeyringCredentialStore` adapter implements this trait.
/// In tests, `TestCredentialProvider` provides a fixed key.
///
/// Note: This mirrors `core::auth::CredentialStore::retrieve()` but is defined
/// here to preserve the leaf-module constraint (cloud/ must not import auth/).
/// The daemon bridges between `auth::CredentialStore` and this trait.
pub trait CredentialProvider: Send + Sync {
    /// Retrieve the current API key. Returns `None` if no credential is available.
    fn retrieve(&self) -> Option<SecretString>;
}

// ---------------------------------------------------------------------------
// CloudConfig — cloud worker configuration
// ---------------------------------------------------------------------------

/// Runtime configuration for the cloud forwarding worker.
///
/// Constructed by the daemon from `config::CloudConfig` fields.
/// Defined here (not in config/) to satisfy the leaf-module constraint.
#[derive(Debug, Clone)]
pub struct CloudConfig {
    /// Cloud API base URL. Default: "https://app.openlatch.ai". Callers
    /// prepend their own `/api/v1/...` path segments.
    pub api_url: String,
    /// TCP connect timeout in milliseconds. Default: 5000.
    pub timeout_connect_ms: u64,
    /// Total request timeout in milliseconds. Default: 10000.
    pub timeout_total_ms: u64,
    /// Delay before the single retry on network error or 5xx, in milliseconds.
    /// Default: 2000.
    pub retry_delay_ms: u64,
    /// Bounded channel size for async event forwarding. Default: 1000.
    pub channel_size: usize,
    /// Default backoff in seconds when Retry-After header is absent. Default: 30.
    pub rate_limit_default_secs: u64,
    /// How often the worker re-reads the credential from the provider. Default:
    /// 60_000 ms (60 s). Overridden to a small value in unit tests so reload
    /// behavior can be exercised without a real-time wait.
    pub credential_poll_interval_ms: u64,
    /// Maximum size of the **unreplayed window** of
    /// `~/.openlatch/logs/fallback.jsonl` — not of the file. Default:
    /// 52_428_800 (50 MB). Setting to 0 disables the cap entirely. See
    /// `config::CloudConfig::fallback_max_bytes` for why the two differ.
    pub fallback_max_bytes: u64,
    /// Maximum number of events accumulated before the worker flushes the
    /// batch. Default: 50. The daemon supplies an already-clamped `1..=100`
    /// value (see `config::Config::load`), so the worker may use it directly:
    /// the platform hard-rejects batches larger than 100, and 0 would leave
    /// the accumulator with no reachable size trigger.
    pub batch_max_events: usize,
    /// Maximum time the worker waits before flushing a partial batch, in
    /// milliseconds. Default: 5000. The deadline is anchored to the *first*
    /// event buffered and is never reset by subsequent ones, so this bounds
    /// forwarding latency instead of merely spacing out flushes.
    pub batch_max_wait_ms: u64,
}

impl Default for CloudConfig {
    fn default() -> Self {
        Self {
            api_url: "https://app.openlatch.ai".into(),
            timeout_connect_ms: 5000,
            timeout_total_ms: 10000,
            retry_delay_ms: 2000,
            channel_size: 1000,
            rate_limit_default_secs: 30,
            credential_poll_interval_ms: 60_000,
            fallback_max_bytes: 50 * 1024 * 1024,
            batch_max_events: 50,
            batch_max_wait_ms: 5000,
        }
    }
}

// ---------------------------------------------------------------------------
// CloudEvent — data passed through the mpsc channel
// ---------------------------------------------------------------------------

/// An event forwarded through the cloud mpsc channel.
///
/// Created by the daemon handler after each verdict and sent via `try_send()`.
/// The envelope is a CloudEvents v1.0.2 structured-mode object (JSON-serialized)
/// — the raw agent payload is already embedded under `envelope.data`, and
/// OpenLatch metadata lives on the envelope as CloudEvents extension attrs.
/// `agent_id` is attached by the worker as the `agentid` extension
/// attribute just before POSTing.
#[derive(Debug, Clone)]
pub struct CloudEvent {
    /// Serialized `EventEnvelope` as a JSON value (already privacy-filtered).
    pub envelope: serde_json::Value,
    /// Agent install identifier from config (`agt_<uuid>`). Empty if not yet
    /// initialized. Stamped by the worker as the CloudEvents `agentid`
    /// extension attribute.
    pub agent_id: String,
}

// ---------------------------------------------------------------------------
// CloudError — errors that can occur during cloud forwarding
// ---------------------------------------------------------------------------

/// Errors returned by the cloud forwarding worker.
#[derive(Debug)]
pub enum CloudError {
    /// Cloud returned 401 or 403 — API key is invalid or revoked.
    AuthError,
    /// Cloud returned 429 — rate limit exceeded; honor the Retry-After delay.
    RateLimit {
        /// Seconds to wait before retrying (parsed from Retry-After header or default).
        retry_after_secs: u64,
    },
    /// Cloud returned a 5xx server error.
    ServerError,
    /// Network error before receiving a response (DNS, connection refused, timeout).
    Network,
    /// Cloud returned an unexpected 4xx client error (not 401, 403, or 429).
    ClientError(u16),
}

impl std::fmt::Display for CloudError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            CloudError::AuthError => {
                write!(f, "cloud auth error (401/403) — API key invalid or revoked")
            }
            CloudError::RateLimit { retry_after_secs } => {
                write!(
                    f,
                    "cloud rate limit (429) — retry after {retry_after_secs}s"
                )
            }
            CloudError::ServerError => write!(f, "cloud server error (5xx)"),
            CloudError::Network => write!(f, "cloud network error — endpoint unreachable"),
            CloudError::ClientError(code) => write!(f, "cloud client error ({code})"),
        }
    }
}

// ---------------------------------------------------------------------------
// CloudState — shared in-process auth error flag
// ---------------------------------------------------------------------------

/// Shared cloud state for in-process visibility of auth error condition and forwarding metrics.
///
/// The `auth_error` atomic flag is read by `openlatch status` (via AppState)
/// to report whether the daemon's cloud forwarding is currently blocked by
/// an auth error. It is also persisted to `cloud_state.json` for cross-process
/// visibility (CLOUD-08).
///
/// `forwarded_count` and `last_sync_secs` are incremented by the worker on
/// each successful 2xx response and exposed via the `/metrics` endpoint.
/// `drop_count` is incremented when an event is dropped after all retries
/// are exhausted (network error, 5xx, or rate-limit retry failure).
#[derive(Debug, Clone)]
pub struct CloudState {
    /// True if the last cloud POST returned 401/403 and the worker is skipping POSTs.
    pub auth_error: Arc<AtomicBool>,
    /// Total number of events successfully forwarded to the cloud (2xx responses).
    pub forwarded_count: Arc<AtomicU64>,
    /// Unix epoch seconds of the last successful cloud forward (0 = never).
    pub last_sync_secs: Arc<AtomicU64>,
    /// Events silently dropped after all retries were exhausted (network/5xx/rate-limit).
    pub drop_count: Arc<AtomicU64>,
    /// Drops since the last successful forward or health check. Drives the live
    /// `cloud_status` classification in the metrics endpoint — unlike `drop_count`,
    /// this is reset to zero on any recovery signal.
    pub consecutive_drops: Arc<AtomicU64>,
    /// True when the cloud worker has no API key and is silently skipping POSTs.
    /// Distinct from `auth_error`: the key has never been loaded (e.g. user not
    /// authenticated), rather than the server rejecting one that was. Reported
    /// as `cloud_status: "no_credential"` so observers don't see a bare
    /// `"connected"` when nothing is actually being forwarded.
    pub no_credential: Arc<AtomicBool>,
    /// Consecutive cloud-health probe failures. Drives `next_health_delay`
    /// exponential backoff: when the cloud is unreachable we poll more
    /// aggressively (5s → 10s → 20s → 30s cap) so `cloud_status` recovers
    /// quickly once connectivity returns, without burning traffic while
    /// healthy (60s). Reset to 0 on any successful probe or successful forward.
    pub consecutive_probe_failures: Arc<AtomicU32>,
    /// Fires whenever the cloud worker confirms connectivity (successful
    /// POST or successful health probe). Both the outbox drain task and the
    /// daemon-side fallback-log replay task listen on this signal so
    /// recovery events captured while offline flow to the cloud as soon as
    /// the first probe succeeds, without waiting for the next hook.
    pub drain_notify: Arc<Notify>,
    /// Live hook events dropped on a full `cloud_tx` since the last reset.
    /// Bumped by the daemon handler's drop path; reset by the worker on
    /// any successful POST. Drives the emergency-mode detector at the
    /// worker's health-tick boundary so a saturated channel triggers
    /// backlog eviction automatically.
    pub consecutive_live_drops: Arc<AtomicU64>,
    /// Unix epoch milliseconds when `consecutive_live_drops` first
    /// incremented from zero. Zeroed alongside the drop counter so a
    /// fresh streak gets a fresh window. The detector compares this to
    /// "now" to require a sustained streak (>10 s by default) before
    /// engaging emergency mode.
    pub live_drops_window_start: Arc<AtomicU64>,
    /// True while the cloud forwarder is in emergency-drop mode. Read by
    /// the fallback-replay task to skip passes entirely (the channel
    /// drains as fast as the worker can post) and exposed in `/metrics`.
    pub emergency_mode: Arc<AtomicBool>,
    /// Unix epoch milliseconds when the cloud-channel depth first crossed
    /// the high-water trip threshold and stayed above the clear threshold.
    /// Zero means depth is currently below the trip threshold (or has dipped
    /// below the clear threshold). The detector compares this to "now" to
    /// engage emergency mode on sustained channel pressure *before* live
    /// events are dropped — sibling signal to `live_drops_window_start`.
    pub channel_high_water_start_ms: Arc<AtomicU64>,
}

impl CloudState {
    /// Create a new `CloudState` with all fields at their initial state.
    pub fn new() -> Self {
        Self {
            auth_error: Arc::new(AtomicBool::new(false)),
            forwarded_count: Arc::new(AtomicU64::new(0)),
            last_sync_secs: Arc::new(AtomicU64::new(0)),
            drop_count: Arc::new(AtomicU64::new(0)),
            consecutive_drops: Arc::new(AtomicU64::new(0)),
            // Start pessimistic: the worker flips this to `false` on its first
            // credential load so `/metrics` reports `no_credential` during the
            // startup window before the first poll has run.
            no_credential: Arc::new(AtomicBool::new(true)),
            consecutive_probe_failures: Arc::new(AtomicU32::new(0)),
            drain_notify: Arc::new(Notify::new()),
            consecutive_live_drops: Arc::new(AtomicU64::new(0)),
            live_drops_window_start: Arc::new(AtomicU64::new(0)),
            emergency_mode: Arc::new(AtomicBool::new(false)),
            channel_high_water_start_ms: Arc::new(AtomicU64::new(0)),
        }
    }

    /// Fire the cloud-recovered signal. Listeners (outbox drain, fallback
    /// replay) wake up and drain pending entries. Idempotent — if nothing is
    /// waiting, the notification is stored and consumed by the next waiter.
    pub fn notify_drain(&self) {
        self.drain_notify.notify_one();
    }

    /// Returns true if the worker currently has no API key.
    pub fn is_no_credential(&self) -> bool {
        self.no_credential.load(Ordering::Relaxed)
    }

    /// Set the `no_credential` flag — called by the worker whenever it polls
    /// the credential provider.
    pub fn set_no_credential(&self, missing: bool) {
        self.no_credential.store(missing, Ordering::Relaxed);
    }

    /// Returns true if there is an active auth error.
    pub fn is_auth_error(&self) -> bool {
        self.auth_error.load(Ordering::Relaxed)
    }

    /// Record a successful cloud forward: increments forwarded_count by 1 and
    /// stores the current Unix epoch seconds in last_sync_secs.
    pub fn record_successful_forward(&self) {
        self.record_successful_forwards(1);
    }

    /// Record `n` successfully forwarded events in one shot.
    ///
    /// The worker POSTs a *batch* of up to `batch_max_events` events per
    /// request, so a per-request `record_successful_forward()` would
    /// under-count `cloud_forwarded_count` by up to 100x on `/metrics` and in
    /// `openlatch status`. Every counter here is per-event, so a batch must
    /// record its length.
    pub fn record_successful_forwards(&self, n: u64) {
        if n == 0 {
            return;
        }
        self.forwarded_count.fetch_add(n, Ordering::Relaxed);
        self.last_sync_secs
            .store(now_unix_secs(), Ordering::Relaxed);
        self.consecutive_drops.store(0, Ordering::Relaxed);
        // The detector owns the transition out of `emergency_mode` (two
        // consecutive clean ticks); here we just feed it a fresh signal.
        self.consecutive_live_drops.store(0, Ordering::Relaxed);
        self.live_drops_window_start.store(0, Ordering::Relaxed);
    }

    /// Record a live hook event dropped on a saturated cloud channel.
    /// Stamps the window start on the first drop of a fresh streak so the
    /// detector can distinguish a transient spike from sustained congestion.
    pub fn record_live_drop(&self) {
        let prev = self.consecutive_live_drops.fetch_add(1, Ordering::Relaxed);
        if prev == 0 {
            self.live_drops_window_start
                .store(now_unix_ms(), Ordering::Relaxed);
        }
    }

    /// Returns the current consecutive-live-drop streak count.
    pub fn consecutive_live_drops(&self) -> u64 {
        self.consecutive_live_drops.load(Ordering::Relaxed)
    }

    /// Returns the Unix epoch milliseconds when the current drop streak
    /// began (0 = no streak active).
    pub fn live_drops_window_start_ms(&self) -> u64 {
        self.live_drops_window_start.load(Ordering::Relaxed)
    }

    /// Returns true while the worker is in emergency drop mode.
    pub fn is_emergency_mode(&self) -> bool {
        self.emergency_mode.load(Ordering::Relaxed)
    }

    /// Update the channel-pressure window based on the latest depth sample.
    /// Stamps `channel_high_water_start_ms` when depth first crosses
    /// `worker::HIGH_WATER_TRIP_PCT`, clears it when depth dips below
    /// `worker::HIGH_WATER_CLEAR_PCT`. Between the two thresholds the
    /// existing stamp is preserved so a brief dip doesn't reset the streak
    /// (hysteresis). Thresholds are detector policy and live with the rest
    /// of the emergency-mode constants in `worker.rs`.
    pub fn note_channel_depth_pct(&self, pct: u64) {
        let current = self.channel_high_water_start_ms.load(Ordering::Relaxed);
        if pct >= worker::HIGH_WATER_TRIP_PCT {
            if current == 0 {
                self.channel_high_water_start_ms
                    .store(now_unix_ms(), Ordering::Relaxed);
            }
        } else if pct < worker::HIGH_WATER_CLEAR_PCT && current != 0 {
            self.channel_high_water_start_ms.store(0, Ordering::Relaxed);
        }
    }

    /// Returns the elapsed milliseconds since channel depth first crossed
    /// the high-water trip threshold, or 0 if depth is currently below it.
    pub fn channel_high_water_window_ms(&self) -> u64 {
        let start = self.channel_high_water_start_ms.load(Ordering::Relaxed);
        if start == 0 {
            0
        } else {
            now_unix_ms().saturating_sub(start)
        }
    }

    /// Set the emergency-mode flag. Owned by the worker's detector — every
    /// other caller in the crate must route through it. Crate-private so
    /// downstream consumers cannot bypass the detector's hysteresis.
    pub(crate) fn set_emergency_mode(&self, on: bool) {
        self.emergency_mode.store(on, Ordering::Relaxed);
    }

    /// Clear the `auth_error` latch out-of-band (e.g. after the CLI confirms
    /// a fresh login via `POST /admin/auth/refresh`). Returns the prior
    /// value so callers can decide whether to persist `cloud_state.json`.
    pub fn clear_auth_error(&self) -> bool {
        self.auth_error.swap(false, Ordering::Relaxed)
    }

    /// Signal that a background health probe against the cloud succeeded.
    /// Clears the recent-drop streak so `cloud_status` can recover without
    /// waiting for the next hook event to arrive. Also resets the probe
    /// backoff so the next health tick returns to the healthy 60s cadence.
    pub fn record_health_ok(&self) {
        self.consecutive_drops.store(0, Ordering::Relaxed);
        self.consecutive_probe_failures.store(0, Ordering::Relaxed);
    }

    /// Record a failed background health probe. Increments the probe-failure
    /// counter that drives `next_health_delay` exponential backoff. Saturates
    /// at `u32::MAX` so a multi-day outage can't wrap around.
    pub fn record_probe_failure(&self) {
        self.consecutive_probe_failures
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_add(1))
            .ok();
    }

    /// Returns the number of consecutive failed health probes since the last
    /// success. Used by tests and metrics.
    pub fn consecutive_probe_failures(&self) -> u32 {
        self.consecutive_probe_failures.load(Ordering::Relaxed)
    }

    /// Delay to sleep before the next cloud health probe.
    ///
    /// - Healthy (0 failures): 60s.
    /// - Degraded (n >= 1 failures): 5s * 2^min(n-1, 3) capped at 30s, then
    ///   scaled by a +/- 20% jitter derived from the current clock's nanosecond
    ///   remainder (good enough to avoid thundering-herd on shared networks
    ///   without pulling in a PRNG crate).
    ///
    /// The schedule is 5s → 10s → 20s → 30s → 30s → … which matches what
    /// production telemetry agents (Datadog, New Relic, Vector) use for
    /// cloud-endpoint reconnect polling.
    pub fn next_health_delay(&self) -> Duration {
        let failures = self.consecutive_probe_failures();
        if failures == 0 {
            return Duration::from_secs(60);
        }
        let exp = (failures - 1).min(3);
        let base_secs = 5u64.saturating_mul(1u64 << exp);
        let capped_secs = base_secs.min(30);
        // Pseudo-random jitter in [80, 120] without a PRNG dep. Good enough —
        // we only need to spread retries across restarting clients, not to
        // resist a statistical attacker.
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.subsec_nanos())
            .unwrap_or(0);
        let jitter_pct = 80 + u64::from(nanos % 41); // 80..=120
        Duration::from_millis(capped_secs * 1000 * jitter_pct / 100)
    }

    /// Returns the current number of successfully forwarded events.
    pub fn forwarded_count(&self) -> u64 {
        self.forwarded_count.load(Ordering::Relaxed)
    }

    /// Returns the Unix epoch seconds of the last successful forward (0 = never).
    pub fn last_sync_secs(&self) -> u64 {
        self.last_sync_secs.load(Ordering::Relaxed)
    }

    /// Increment the drop counter — called whenever an event is silently
    /// discarded after all retries are exhausted.
    pub fn record_drop(&self) {
        self.record_drops(1);
    }

    /// Record `n` dropped events in one shot. A batch of `n` events that
    /// exhausts its retries is `n` drops, not one — the counter is per-event
    /// everywhere it is read.
    pub fn record_drops(&self, n: u64) {
        if n == 0 {
            return;
        }
        self.drop_count.fetch_add(n, Ordering::Relaxed);
        self.consecutive_drops.fetch_add(n, Ordering::Relaxed);
    }

    /// Returns the cumulative number of dropped events since daemon start.
    pub fn drop_count(&self) -> u64 {
        self.drop_count.load(Ordering::Relaxed)
    }

    /// Returns drops since the last successful forward or health probe.
    pub fn consecutive_drops(&self) -> u64 {
        self.consecutive_drops.load(Ordering::Relaxed)
    }
}

impl Default for CloudState {
    fn default() -> Self {
        Self::new()
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_cloud_config_default_produces_expected_values() {
        let cfg = CloudConfig::default();
        assert_eq!(cfg.api_url, "https://app.openlatch.ai");
        assert_eq!(cfg.timeout_connect_ms, 5000);
        assert_eq!(cfg.timeout_total_ms, 10000);
        assert_eq!(cfg.channel_size, 1000);
        assert_eq!(cfg.retry_delay_ms, 2000);
        assert_eq!(cfg.rate_limit_default_secs, 30);
        assert_eq!(cfg.credential_poll_interval_ms, 60_000);
        assert_eq!(cfg.fallback_max_bytes, 52_428_800);
    }

    #[test]
    fn test_cloud_error_variants_have_display() {
        // AuthError
        let e = CloudError::AuthError;
        assert!(format!("{e}").contains("auth error"));

        // RateLimit
        let e = CloudError::RateLimit {
            retry_after_secs: 42,
        };
        let s = format!("{e}");
        assert!(s.contains("rate limit") && s.contains("42"));

        // ServerError
        let e = CloudError::ServerError;
        assert!(format!("{e}").contains("server error"));

        // Network
        let e = CloudError::Network;
        assert!(format!("{e}").contains("network"));

        // ClientError
        let e = CloudError::ClientError(404);
        let s = format!("{e}");
        assert!(s.contains("404"));
    }

    #[test]
    fn test_cloud_state_defaults_to_no_auth_error() {
        let state = CloudState::new();
        assert!(!state.is_auth_error());
    }

    #[test]
    fn test_cloud_state_reflects_atomic_transitions() {
        let state = CloudState::new();
        state.auth_error.store(true, Ordering::Relaxed);
        assert!(state.is_auth_error());
        state.auth_error.store(false, Ordering::Relaxed);
        assert!(!state.is_auth_error());
    }

    #[test]
    fn test_cloud_state_no_credential_defaults_true_and_toggles() {
        let state = CloudState::new();
        assert!(
            state.is_no_credential(),
            "new CloudState must start in no_credential state until the worker polls"
        );
        state.set_no_credential(false);
        assert!(!state.is_no_credential());
        state.set_no_credential(true);
        assert!(state.is_no_credential());
    }

    #[test]
    fn test_cloud_event_fields_accessible() {
        let evt = CloudEvent {
            envelope: serde_json::json!({"id": "evt_123"}),
            agent_id: "agt_abc".to_string(),
        };
        assert_eq!(evt.agent_id, "agt_abc");
        assert_eq!(evt.envelope["id"], "evt_123");
    }

    #[test]
    fn test_cloud_state_new_initializes_forwarded_count_to_zero() {
        let state = CloudState::new();
        assert_eq!(state.forwarded_count(), 0);
    }

    #[test]
    fn test_cloud_state_new_initializes_last_sync_secs_to_zero() {
        let state = CloudState::new();
        assert_eq!(state.last_sync_secs(), 0);
    }

    #[test]
    fn test_cloud_state_record_successful_forward_increments_count() {
        let state = CloudState::new();
        state.record_successful_forward();
        assert_eq!(state.forwarded_count(), 1);
        state.record_successful_forward();
        assert_eq!(state.forwarded_count(), 2);
    }

    #[test]
    fn test_next_health_delay_healthy_is_sixty_seconds() {
        let state = CloudState::new();
        assert_eq!(state.next_health_delay(), Duration::from_secs(60));
    }

    #[test]
    fn test_next_health_delay_degraded_in_expected_range() {
        let state = CloudState::new();
        // 1 failure -> 5s * jitter [80..=120%] = 4000..=6000 ms
        state.record_probe_failure();
        let d = state.next_health_delay();
        assert!(d >= Duration::from_millis(4000) && d <= Duration::from_millis(6000));

        // Escalate to 4 failures -> 30s capped * jitter = 24000..=36000 ms
        for _ in 0..3 {
            state.record_probe_failure();
        }
        let d = state.next_health_delay();
        assert!(d >= Duration::from_millis(24_000) && d <= Duration::from_millis(36_000));
    }

    #[test]
    fn test_record_live_drop_increments_and_stamps_window_start() {
        let state = CloudState::new();
        assert_eq!(state.consecutive_live_drops(), 0);
        assert_eq!(state.live_drops_window_start_ms(), 0);
        state.record_live_drop();
        assert_eq!(state.consecutive_live_drops(), 1);
        let window_start = state.live_drops_window_start_ms();
        assert!(
            window_start > 0,
            "window start must be stamped on first drop"
        );
        // Subsequent drops keep the original window start.
        state.record_live_drop();
        state.record_live_drop();
        assert_eq!(state.consecutive_live_drops(), 3);
        assert_eq!(state.live_drops_window_start_ms(), window_start);
    }

    #[test]
    fn test_record_successful_forward_clears_live_drop_streak() {
        let state = CloudState::new();
        state.record_live_drop();
        state.record_live_drop();
        assert_eq!(state.consecutive_live_drops(), 2);
        state.record_successful_forward();
        assert_eq!(state.consecutive_live_drops(), 0);
        assert_eq!(state.live_drops_window_start_ms(), 0);
    }

    #[test]
    fn test_emergency_mode_flag_round_trip() {
        let state = CloudState::new();
        assert!(!state.is_emergency_mode());
        state.set_emergency_mode(true);
        assert!(state.is_emergency_mode());
        state.set_emergency_mode(false);
        assert!(!state.is_emergency_mode());
    }

    #[test]
    fn test_note_channel_depth_pct_hysteresis() {
        let state = CloudState::new();
        // Below trip → no streak.
        state.note_channel_depth_pct(state_pct_below_trip());
        assert_eq!(state.channel_high_water_window_ms(), 0);
        // Crosses trip → stamp.
        state.note_channel_depth_pct(state_pct_above_trip());
        let first_start = state.channel_high_water_start_ms.load(Ordering::Relaxed);
        assert!(first_start > 0);
        // In the hysteresis band (above clear but below trip) → preserve stamp.
        state.note_channel_depth_pct(state_pct_in_band());
        assert_eq!(
            state.channel_high_water_start_ms.load(Ordering::Relaxed),
            first_start
        );
        // Below clear → clear stamp.
        state.note_channel_depth_pct(state_pct_below_clear());
        assert_eq!(state.channel_high_water_start_ms.load(Ordering::Relaxed), 0);
        assert_eq!(state.channel_high_water_window_ms(), 0);
    }

    fn state_pct_below_trip() -> u64 {
        worker::HIGH_WATER_TRIP_PCT.saturating_sub(10)
    }
    fn state_pct_above_trip() -> u64 {
        worker::HIGH_WATER_TRIP_PCT + 1
    }
    fn state_pct_in_band() -> u64 {
        (worker::HIGH_WATER_TRIP_PCT + worker::HIGH_WATER_CLEAR_PCT) / 2
    }
    fn state_pct_below_clear() -> u64 {
        worker::HIGH_WATER_CLEAR_PCT.saturating_sub(10)
    }

    #[test]
    fn test_record_health_ok_resets_probe_failures() {
        let state = CloudState::new();
        state.record_probe_failure();
        state.record_probe_failure();
        assert_eq!(state.consecutive_probe_failures(), 2);
        state.record_health_ok();
        assert_eq!(state.consecutive_probe_failures(), 0);
    }

    #[test]
    fn test_cloud_state_record_successful_forward_sets_last_sync_secs() {
        let before = now_unix_secs();
        let state = CloudState::new();
        state.record_successful_forward();
        let after = now_unix_secs();
        let recorded = state.last_sync_secs();
        assert!(
            recorded >= before,
            "last_sync_secs should be >= before: {recorded} < {before}"
        );
        assert!(
            recorded <= after,
            "last_sync_secs should be <= after: {recorded} > {after}"
        );
    }
}