openlatch-client 0.5.8

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

    /// Drop any process-local cache backing [`retrieve`](Self::retrieve), so the
    /// next call re-reads the underlying store instead of a memoized answer.
    ///
    /// Default no-op — the test providers never cache. The production adapter
    /// (`daemon::CredentialStoreAdapter`) delegates to
    /// `core::auth::CredentialStore::invalidate`, which is where the OS
    /// keychain's process-lifetime memo actually lives (`cloud/` must not
    /// import `auth/`, hence the indirection). Called when this process learns
    /// a credential changed out of band: the `/admin/auth/refresh` route and
    /// the worker's own 401/403 latch.
    fn invalidate(&self) {}
}

// ---------------------------------------------------------------------------
// 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,
    /// The `X-OpenLatch-Machine-Id` value sent on every cloud request: the
    /// host key this install reports for licensing — its `hostid`, or
    /// `agent:<agent_id>` for a host with no machine identifier. `None` omits
    /// the header entirely, which is what a host with neither must do; the
    /// platform's host-key pattern rejects a bare `agent:`.
    ///
    /// Computed once by the daemon (`daemon::identity::host_key`) and carried
    /// here as a value: `core` never reaches into `crate::daemon`.
    pub host_key: Option<String>,
    /// The `hostid` that [`worker::stamp_extensions`](crate::cloud) fills into
    /// an envelope carrying none — the config monitor's, the tamper
    /// reconciler's and the model relay's, which never pass through
    /// `daemon/handlers.rs`. `None` fills nothing.
    ///
    /// Deliberately **not** `host_key`: that field's `agent:<agent_id>`
    /// fallback is a licensing key, not a machine identifier, and stamping it
    /// as `hostid` would put a value on the wire that violates the attribute's
    /// declared shape.
    pub host_id: Option<String>,
    /// The `X-OpenLatch-Agent-Id` value sent on every ingest request: this
    /// install's persistent id from `config.toml` — the same value the policy
    /// poller and the config-monitor alerts long-poll already send under that
    /// name. `None` before provisioning omits the header entirely rather than
    /// sending it empty. The platform uses it only when an ingested envelope
    /// carries no `agentid` (openlatch-platform#915), which is exactly the
    /// case for the empty-batch renegotiation probe — it carries no envelope
    /// at all, so without this header its acknowledgement lands attributed to
    /// no install.
    pub agent_id: Option<String>,
}

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,
            host_key: None,
            host_id: None,
            agent_id: None,
        }
    }
}

// ---------------------------------------------------------------------------
// 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,
    },
    /// The platform refused this batch on **licensing** grounds — a `402`
    /// (whatever its body), or a `503` whose body code is a site-licence one.
    ///
    /// **Never a credential failure, and never a reason to stop.** The batch is
    /// retained — spooled to the outbox, or counted as drops only when there is
    /// no outbox — and retried on the license cadence; the resident policy
    /// bundle keeps deciding throughout. A 402 used to fall into
    /// [`CloudError::ClientError`], where the batch was dropped as terminal.
    LicenseRefused {
        /// The platform's `error.code`, or `license_unknown` when the body
        /// could not be parsed (402 is RFC-undefined; retaining is the safe
        /// default).
        code: String,
        /// Where a human goes to fix it, when the platform named one.
        licensing_url: Option<String>,
        /// The `Retry-After` the gate asked for, already capped at
        /// [`MAX_RETRY_AFTER_SECS`].
        retry_after: Option<Duration>,
    },
    /// 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, 402, or 429).
    ClientError(u16),
    /// Negotiation has no safe event representation. Retain and renegotiate;
    /// never classify this as auth or as a disposable generic 4xx.
    CompatibilityUnavailable,
}

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::LicenseRefused { code, .. } => {
                write!(f, "cloud license refusal ({code}) — events are queued")
            }
            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})"),
            CloudError::CompatibilityUnavailable => {
                write!(f, "cloud protocol compatibility unavailable")
            }
        }
    }
}

/// Upper bound applied to any server-supplied `Retry-After`.
///
/// An origin — or something impersonating one on a hostile network — must not
/// be able to park a client for a week with one header. Shared by the cloud
/// worker's license gate and the policy poller, which is why it lives here
/// rather than in either of them: `src/core` never imports `crate::daemon`, so
/// the poller imports this and not the other way round.
pub const MAX_RETRY_AFTER_SECS: u64 = 3_600;

/// The license cadence when the gate names no `Retry-After`: 15 minutes.
///
/// A licence does not un-expire in thirty seconds, and the ordinary retry
/// ladder would spend the life of an expired contract logging.
pub const LICENSE_RETRY_SECS: u64 = 900;

/// Parse a `Retry-After` value into whole seconds from now.
///
/// RFC 9110 §10.2.3 allows **two** forms and a rate-limiting origin may send
/// either:
///
/// * `delta-seconds` — `Retry-After: 120`
/// * `HTTP-date` — `Retry-After: Wed, 21 Oct 2026 07:28:00 GMT`
///
/// Parsing only the first means a date-form header is silently ignored and the
/// caller retries at its ordinary interval instead — hitting an origin that
/// just told us it is overloaded, sooner than it asked. A date already in the
/// past yields `0`, i.e. "retry at the next ordinary tick", never a negative
/// that would wrap the unsigned conversion.
pub(crate) fn parse_retry_after(raw: &str) -> Option<u64> {
    let value = raw.trim();
    if let Ok(secs) = value.parse::<u64>() {
        return Some(secs);
    }
    let when = chrono::DateTime::parse_from_rfc2822(value).ok()?;
    let delta = when.timestamp() - chrono::Utc::now().timestamp();
    Some(delta.max(0) as u64)
}

/// The three site-level codes a `503` may carry, which are licence refusals
/// rather than server faults.
///
/// `clock_regression` is one of them because a site licence is time-bounded: a
/// host whose clock has gone backwards cannot be told apart from one outside
/// its term, and either way the answer is "wait", not "retry hard".
pub(crate) const SITE_LICENSE_CODES: [&str; 3] = [
    "site_license_missing",
    "site_license_expired",
    "clock_regression",
];

/// `error.code` out of the platform's house error envelope, when the body is
/// one and carries a code.
///
/// One reader for all of them: the cloud worker, the policy poller's `503` arm,
/// the hold answer source and `openlatch status` all read the same field out of
/// the same envelope, and four copies of one JSON pointer is four places for
/// the path to drift.
pub(crate) fn error_code_of(body: &str) -> Option<String> {
    serde_json::from_str::<serde_json::Value>(body)
        .ok()
        .as_ref()
        .and_then(|v| v.pointer("/error/code"))
        .and_then(serde_json::Value::as_str)
        .map(str::to_string)
}

/// True when a house-envelope body's `error.code` is a site-licence code.
///
/// Read by `post_batch` (and so by the outbox drain), by the policy poller's
/// `503` arm and by the hold answer source, so that a clock-regressed site
/// never burns outbox attempts on any of the three paths.
pub(crate) fn body_code_is_site_license(body: &str) -> bool {
    error_code_of(body).is_some_and(|code| SITE_LICENSE_CODES.contains(&code.as_str()))
}

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

/// A live licence refusal, and when it is worth POSTing again.
///
/// Held in memory only. It is deliberately **not** persisted to
/// `cloud_state.json` — that file carries `{auth_error, updated_at}` and
/// nothing else, and a refusal that survived a restart would keep a host
/// silent after the contract it complained about was renewed. A restart costs
/// one refused batch; a stale persisted gate costs the whole backlog.
#[derive(Debug, Clone)]
pub struct LicenseGate {
    /// The platform's `error.code` — `license_expired`, `site_license_missing`,
    /// `license_unknown` for a body that would not parse, and so on.
    pub code: String,
    /// Where a human goes to fix it, when the platform named one.
    pub licensing_url: Option<String>,
    /// Wall-clock time the refusal was seen, for `/metrics` and `status`.
    pub at: std::time::SystemTime,
    /// The instant after which a POST is worth attempting again.
    pub until: std::time::Instant,
}

/// 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), waking the daemon-side fallback-log
    /// replay task so entries captured while offline flow to the cloud as
    /// soon as the first probe succeeds, without waiting for the next hook.
    ///
    /// Deliberately a *separate* `Notify` from `outbox_drain_notify` rather
    /// than one shared handle: `Notify::notify_one()` wakes exactly one
    /// waiter and stores at most one permit, so two independent consumers
    /// cloning the same `Notify` would race — one recovery generation could
    /// wake only whichever task happened to be polling. One `Notify` per
    /// consumer keeps the "at most one waiter, at most one permit" contract
    /// tokio actually provides: a signal that lands before that consumer
    /// starts waiting is still stored and consumed by the very next
    /// `notified().await` (see `notify_drain()`).
    pub fallback_drain_notify: Arc<Notify>,
    /// Sibling of `fallback_drain_notify` for the outbox drain task. See
    /// that field's doc for why this is a dedicated `Notify` rather than a
    /// shared one.
    pub outbox_drain_notify: Arc<Notify>,
    /// Coalesced cross-family refresh request. Event-ingest incompatibility
    /// wakes the policy poller independently; it never waits for bundle-family
    /// compatibility before renegotiating the event writer.
    pub policy_refresh_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 model relay 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>,
    /// The live licence refusal, if any (I-4).
    ///
    /// `Some` means the platform has refused ingest on licensing grounds and no
    /// POST has succeeded since — the live path spools without POSTing while
    /// `until` is in the future, and `/metrics` reports `license_blocked` until
    /// a successful POST clears it. An `ArcSwapOption` rather than a `Mutex`
    /// because this is read on every flush and written once per refusal.
    pub license_gate: Arc<arc_swap::ArcSwapOption<LicenseGate>>,
    /// The policy poller's live licence refusal — the bundle-side twin of
    /// `license_gate`, set when the bundle fetch is refused on licensing
    /// grounds and cleared when a bundle activates.
    ///
    /// Kept apart from `license_gate` on purpose: a refused bundle must not
    /// hold ingest POSTs, and the Cloud row must not claim events are queued
    /// when none was refused. Its reader is the doctor's Policy row, which
    /// would otherwise only see an unacknowledged compatibility selection on an
    /// organization that has simply never been licensed.
    pub policy_license_refusal: Arc<arc_swap::ArcSwapOption<LicenseGate>>,
}

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)),
            fallback_drain_notify: Arc::new(Notify::new()),
            outbox_drain_notify: Arc::new(Notify::new()),
            policy_refresh_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)),
            license_gate: Arc::new(arc_swap::ArcSwapOption::empty()),
            policy_license_refusal: Arc::new(arc_swap::ArcSwapOption::empty()),
        }
    }

    /// Fire the cloud-recovered signal. Every listener (outbox drain,
    /// fallback replay) wakes and drains pending entries — each has its own
    /// `Notify`, so one recovery generation reaches both rather than
    /// whichever one happened to win a shared wait. Idempotent per listener:
    /// if a listener isn't waiting yet, its notification is stored and
    /// consumed by that listener's next wait.
    pub fn notify_drain(&self) {
        self.fallback_drain_notify.notify_one();
        self.outbox_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)
    }

    /// Record a licence refusal and hold POSTs until the cadence expires.
    ///
    /// `retry_after` is what the gate asked for, already capped; without one
    /// the wait is [`LICENSE_RETRY_SECS`].
    pub fn set_license_gate(
        &self,
        code: String,
        licensing_url: Option<String>,
        retry_after: Option<Duration>,
    ) {
        let wait = retry_after.unwrap_or(Duration::from_secs(LICENSE_RETRY_SECS));
        self.license_gate.store(Some(Arc::new(LicenseGate {
            code,
            licensing_url,
            at: std::time::SystemTime::now(),
            until: std::time::Instant::now() + wait,
        })));
    }

    /// True while a gate exists **and** its cadence has not elapsed — i.e. a
    /// POST right now would only re-earn the same refusal.
    ///
    /// Deliberately narrower than [`CloudState::license_gate`]: once `until`
    /// passes, the next POST is due and this returns `false`, but the gate
    /// itself stays until that POST actually succeeds, which is what keeps
    /// `/metrics` honest in the meantime.
    pub fn license_gate_active(&self) -> bool {
        self.license_gate
            .load()
            .as_ref()
            .is_some_and(|gate| gate.until > std::time::Instant::now())
    }

    /// The live gate, for `/metrics` and `openlatch status`.
    pub fn license_gate(&self) -> Option<Arc<LicenseGate>> {
        self.license_gate.load_full()
    }

    /// Clear the gate. Returns whether one was there, so the caller can log
    /// the recovery exactly once.
    pub fn clear_license_gate(&self) -> bool {
        self.license_gate.swap(None).is_some()
    }

    /// Record (`Some`) or clear (`None`) the policy poller's licence refusal.
    /// The poller schedules its own retry, so `until` is not consulted.
    pub fn set_policy_license_refusal(&self, refusal: Option<(String, Option<String>)>) {
        self.policy_license_refusal
            .store(refusal.map(|(code, licensing_url)| {
                Arc::new(LicenseGate {
                    code,
                    licensing_url,
                    at: std::time::SystemTime::now(),
                    until: std::time::Instant::now(),
                })
            }));
    }

    /// The policy poller's live licence refusal, for `/metrics`.
    pub fn policy_license_refusal(&self) -> Option<Arc<LicenseGate>> {
        self.policy_license_refusal.load_full()
    }

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

    // -----------------------------------------------------------------------
    // Retry-After (RFC 9110 §10.2.3 — BOTH forms)
    //
    // Moved here with `parse_retry_after` when the license gate began sharing
    // it with the policy poller; `src/core` never imports `crate::daemon`, so
    // the parser and its tests live on this side of the boundary.
    // -----------------------------------------------------------------------

    fn susp_range() -> std::ops::RangeInclusive<u64> {
        590..=600
    }

    #[test]
    fn retry_after_parses_delta_seconds() {
        assert_eq!(parse_retry_after("120"), Some(120));
        assert_eq!(parse_retry_after("  120  "), Some(120));
    }

    #[test]
    fn retry_after_parses_the_http_date_form() {
        // The date form is legal and rate-limiting origins do send it. Ignoring
        // it means retrying at the ordinary interval against an origin that
        // just said it was overloaded.
        let future = chrono::Utc::now() + chrono::Duration::seconds(600);
        let header = future.format("%a, %d %b %Y %H:%M:%S GMT").to_string();
        let parsed = parse_retry_after(&header).expect("http-date must parse");
        assert!(
            (susp_range()).contains(&parsed),
            "expected ~600s from an http-date, got {parsed}"
        );
    }

    #[test]
    fn retry_after_in_the_past_is_zero_not_a_wrapped_negative() {
        let past = chrono::Utc::now() - chrono::Duration::seconds(600);
        let header = past.format("%a, %d %b %Y %H:%M:%S GMT").to_string();
        assert_eq!(parse_retry_after(&header), Some(0));
    }

    #[test]
    fn retry_after_garbage_is_none_not_a_panic() {
        assert_eq!(parse_retry_after("soon"), None);
        assert_eq!(parse_retry_after(""), None);
        assert_eq!(parse_retry_after("-5"), None);
    }

    // -----------------------------------------------------------------------
    // Site-licence body codes (I-4)
    // -----------------------------------------------------------------------

    /// A `503` is a server fault unless its body names a site-licence code —
    /// that distinction is what decides between "retry hard" and "wait".
    #[test]
    fn only_the_three_site_codes_make_a_503_a_licence_refusal() {
        for code in [
            "site_license_missing",
            "site_license_expired",
            "clock_regression",
        ] {
            let body = format!(r#"{{"error":{{"code":"{code}","message":"nope"}}}}"#);
            assert!(body_code_is_site_license(&body), "{code} must be a refusal");
        }

        // `license_expired` is an ORG-level code and rides a 402; on a 503 it
        // is not one of the three and the batch is a server error.
        assert!(!body_code_is_site_license(
            r#"{"error":{"code":"license_expired"}}"#
        ));
        assert!(!body_code_is_site_license(r#"{"error":{}}"#));
        assert!(!body_code_is_site_license("{}"));
        assert!(!body_code_is_site_license(""));
        assert!(!body_code_is_site_license("<html>502 Bad Gateway</html>"));
    }

    // -----------------------------------------------------------------------
    // The license gate (I-4)
    // -----------------------------------------------------------------------

    /// The gate's two questions are deliberately different: "is a POST due?"
    /// (`license_gate_active`) stops being true the moment the cadence
    /// elapses, while "has a POST proved the refusal is over?"
    /// (`license_gate`) stays false until one actually succeeds. Collapsing
    /// them would make an idle host report itself healthy without ever having
    /// reached the platform.
    #[test]
    fn a_gate_stops_holding_posts_before_it_stops_being_reported() {
        let state = CloudState::new();
        assert!(!state.license_gate_active());
        assert!(state.license_gate().is_none());

        state.set_license_gate(
            "license_expired".to_string(),
            Some("https://app.openlatch.ai/settings/licensing".to_string()),
            Some(Duration::from_secs(900)),
        );
        assert!(state.license_gate_active(), "a fresh gate holds POSTs");
        let gate = state.license_gate().expect("the gate is reportable");
        assert_eq!(gate.code, "license_expired");
        assert_eq!(
            gate.licensing_url.as_deref(),
            Some("https://app.openlatch.ai/settings/licensing")
        );

        // Cadence elapsed: the next POST is due, but nothing has proved the
        // refusal is over, so the gate is still reported.
        state.set_license_gate("license_expired".to_string(), None, Some(Duration::ZERO));
        assert!(!state.license_gate_active());
        assert!(state.license_gate().is_some());

        assert!(state.clear_license_gate(), "clearing reports it was set");
        assert!(state.license_gate().is_none());
        assert!(
            !state.clear_license_gate(),
            "clearing an absent gate is not a recovery to log"
        );
    }

    /// No `Retry-After` means the 15-minute fallback, not the next tick.
    #[test]
    fn a_gate_with_no_retry_after_waits_the_license_cadence() {
        let state = CloudState::new();
        state.set_license_gate("license_unknown".to_string(), None, None);
        let gate = state.license_gate().expect("gate");
        let wait = gate
            .until
            .saturating_duration_since(std::time::Instant::now());
        assert!(
            wait > Duration::from_secs(LICENSE_RETRY_SECS - 5)
                && wait <= Duration::from_secs(LICENSE_RETRY_SECS),
            "expected ~{LICENSE_RETRY_SECS}s, got {wait:?}"
        );
    }

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

    /// Regression for #307: `notify_drain()` must wake *every* consumer
    /// waiting on the recovery signal, not just whichever one happened to
    /// be the single `Notify::notify_one()` waiter.
    ///
    /// Reproduces the real topology — fallback replay and outbox drain each
    /// clone their own dedicated `Notify` and loop on `notified().await` —
    /// and asserts both tasks observe one recovery generation. Bounded by a
    /// timeout so a regression fails as a clean assertion, not a hang.
    #[tokio::test]
    async fn test_notify_drain_wakes_every_consumer() {
        let state = CloudState::new();

        let fallback_notify = state.fallback_drain_notify.clone();
        let outbox_notify = state.outbox_drain_notify.clone();

        let fallback_task = tokio::spawn(async move {
            fallback_notify.notified().await;
        });
        let outbox_task = tokio::spawn(async move {
            outbox_notify.notified().await;
        });

        // Give both tasks a chance to reach `notified().await` before the
        // single recovery signal fires.
        tokio::task::yield_now().await;
        tokio::task::yield_now().await;

        state.notify_drain();

        tokio::time::timeout(Duration::from_secs(2), async {
            fallback_task.await.expect("fallback task panicked");
            outbox_task.await.expect("outbox task panicked");
        })
        .await
        .expect("one notify_drain() signal must wake both consumers");
    }

    /// Second acceptance line: a recovery signal that lands *before* a
    /// consumer starts waiting must not be lost.
    #[tokio::test]
    async fn test_notify_drain_signal_before_wait_is_not_lost() {
        let state = CloudState::new();

        // Fire before either consumer is waiting.
        state.notify_drain();

        let fallback_notify = state.fallback_drain_notify.clone();
        let outbox_notify = state.outbox_drain_notify.clone();

        let fallback_task = tokio::spawn(async move {
            fallback_notify.notified().await;
        });
        let outbox_task = tokio::spawn(async move {
            outbox_notify.notified().await;
        });

        tokio::time::timeout(Duration::from_secs(2), async {
            fallback_task.await.expect("fallback task panicked");
            outbox_task.await.expect("outbox task panicked");
        })
        .await
        .expect("a signal that landed before either consumer waited must not be dropped");
    }

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