openlatch-client 0.1.7

//! 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>`
//! - Processes events sequentially — one POST per event (D-16)
//! - 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::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

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

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

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

/// Background cloud health probe interval. On success the worker clears the
/// `consecutive_drops` counter so `cloud_status` recovers without waiting for
/// the next hook event.
#[cfg(not(test))]
const CLOUD_HEALTH_INTERVAL: Duration = Duration::from_secs(60);
#[cfg(test)]
const CLOUD_HEALTH_INTERVAL: Duration = Duration::from_millis(50);

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

/// Main cloud worker loop.
///
/// Consumes events from `rx` and POSTs them 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).
///
/// # Parameters
///
/// - `rx`: receiver end of the bounded mpsc channel
/// - `credential_provider`: source of the current API key (hot-reloaded every 60s)
/// - `config`: cloud forwarding configuration
/// - `cloud_state`: shared in-process auth error flag (readable by AppState/status command)
/// - `openlatch_dir`: base directory for writing `cloud_state.json`
pub async fn run_cloud_worker(
    mut rx: mpsc::Receiver<CloudEvent>,
    credential_provider: Arc<dyn CredentialProvider>,
    config: CloudConfig,
    cloud_state: CloudState,
    openlatch_dir: PathBuf,
) {
    let client = build_cloud_client(&config);
    let credential_poll_interval = Duration::from_millis(config.credential_poll_interval_ms);

    // 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;
    // 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();
    if current_key.is_none() {
        tracing::warn!(
            code = "OL-1200",
            "cloud worker: no API key available — cloud forwarding disabled until you run \
             'openlatch auth login' or set [cloud] enabled = false in config.toml (fail-open, \
             events are still logged locally)"
        );
    }
    let mut last_credential_poll = tokio::time::Instant::now();
    let mut last_health_tick = tokio::time::Instant::now();

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

            // Reset auth_error whenever a new key is present — even if it is the same
            // value — so that re-running `openlatch auth login` with the same revoked key
            // clears the error flag and lets the next POST determine validity (WR-01).
            if new_key.is_some() {
                if key_changed {
                    tracing::info!(
                        "cloud worker: credential refreshed — resetting auth_error state"
                    );
                }
                if auth_error {
                    auth_error = false;
                    cloud_state
                        .auth_error
                        .store(false, std::sync::atomic::Ordering::Relaxed);
                    if let Err(e) = persist_cloud_state(&openlatch_dir, false) {
                        tracing::warn!(error = %e, "cloud worker: failed to persist cloud_state.json on credential refresh");
                    }
                }
                // A credential is now present — arm the OL-1200 warning latch so
                // a future missing-key streak will warn once again.
                missing_key_warned = false;
                current_key = new_key;
            }

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

        // Receive next event (with timeout to allow periodic credential polling)
        let event = tokio::select! {
            maybe = rx.recv() => {
                match maybe {
                    Some(event) => event,
                    None => {
                        // Channel closed — all senders dropped, daemon is shutting down
                        tracing::info!("cloud worker: channel closed, exiting");
                        return;
                    }
                }
            }
            // 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.
            _ = tokio::time::sleep_until(
                last_health_tick + CLOUD_HEALTH_INTERVAL
            ) => {
                last_health_tick = tokio::time::Instant::now();
                match cloud_health_check(&client, &config).await {
                    Ok(()) => cloud_state.record_health_ok(),
                    Err(e) => tracing::debug!(error = %e, "cloud health check failed"),
                }
                continue;
            }
        };

        // Auth error state: drain channel without POSTing until credential refreshes (D-14)
        if auth_error {
            tracing::debug!(
                code = "OL-1201",
                "cloud worker: auth_error active — skipping POST for event"
            );
            continue;
        }

        // Need a valid key to POST
        let key = match &current_key {
            Some(k) => k.clone(),
            None => {
                if !missing_key_warned {
                    tracing::warn!(
                        code = "OL-1200",
                        "cloud worker: no API key available — cloud forwarding disabled until \
                         you run 'openlatch auth login' or set [cloud] enabled = false in \
                         config.toml (fail-open, events are still logged locally)"
                    );
                    missing_key_warned = true;
                } else {
                    tracing::debug!(
                        code = "OL-1200",
                        "cloud worker: skipping event — still no API key available"
                    );
                }
                continue;
            }
        };

        // Attempt POST (with one retry on network error or 5xx)
        match post_event(&client, &config, &key, &event, &openlatch_dir).await {
            Ok(()) => {
                tracing::debug!("cloud worker: event forwarded successfully");
                cloud_state.record_successful_forward();
            }
            Err(CloudError::AuthError) => {
                tracing::warn!(
                    code = "OL-1201",
                    "cloud worker: auth error (401/403) — pausing POSTs until credential refresh"
                );
                auth_error = true;
                cloud_state
                    .auth_error
                    .store(true, std::sync::atomic::Ordering::Relaxed);
                if let Err(e) = persist_cloud_state(&openlatch_dir, true) {
                    tracing::warn!(error = %e, "cloud worker: failed to persist cloud_state.json");
                }
            }
            Err(CloudError::RateLimit { retry_after_secs }) => {
                tracing::warn!(
                    code = "OL-1202",
                    retry_after_secs,
                    "cloud worker: rate limited (429) — sleeping then retrying once"
                );
                tokio::time::sleep(Duration::from_secs(retry_after_secs)).await;
                // Retry once after the wait
                if let Err(e) = post_event(&client, &config, &key, &event, &openlatch_dir).await {
                    tracing::warn!(
                        error = %e,
                        "cloud worker: event dropped after rate-limit retry"
                    );
                    cloud_state.record_drop();
                }
            }
            Err(CloudError::ServerError) => {
                tracing::warn!(
                    code = "OL-1200",
                    "cloud worker: server error (5xx) — retrying once after delay"
                );
                tokio::time::sleep(Duration::from_millis(config.retry_delay_ms)).await;
                if let Err(e) = post_event(&client, &config, &key, &event, &openlatch_dir).await {
                    tracing::warn!(error = %e, "cloud worker: event dropped after 5xx retry");
                    cloud_state.record_drop();
                }
            }
            Err(CloudError::Network) => {
                tracing::warn!(
                    code = "OL-1200",
                    "cloud worker: network error — retrying once after delay"
                );
                tokio::time::sleep(Duration::from_millis(config.retry_delay_ms)).await;
                if let Err(e) = post_event(&client, &config, &key, &event, &openlatch_dir).await {
                    tracing::warn!(error = %e, "cloud worker: event dropped after network retry");
                    cloud_state.record_drop();
                }
            }
            Err(CloudError::ClientError(code)) => {
                tracing::warn!(
                    http_status = code,
                    "cloud worker: unexpected 4xx — dropping event (no retry)"
                );
            }
        }
    }
}

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

/// POST a single cloud event to the ingestion endpoint.
///
/// Builds the `CloudIngestionRequest` body, attaches headers with UUIDv7 request ID,
/// and interprets the HTTP response status.
///
/// # 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_event(
    client: &reqwest::Client,
    config: &CloudConfig,
    key: &SecretString,
    event: &CloudEvent,
    _openlatch_dir: &Path,
) -> Result<(), CloudError> {
    // Generate UUIDv7 request ID (D-19)
    let request_id = uuid::Uuid::now_v7().to_string();

    // Stamp the CloudEvents `agentid` extension attribute onto the envelope
    // just before serialisation, then POST the single-event batch per
    // CloudEvents v1.0.2 structured-mode batch format.
    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()),
        );
    }
    let body = serde_json::Value::Array(vec![envelope]);

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

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

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

    let status = response.status();

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

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

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

    let updated_at = {
        // RFC3339 UTC timestamp using std only (no chrono dependency in this leaf module)
        use std::time::{SystemTime, UNIX_EPOCH};
        let secs = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();
        // Format as ISO 8601 UTC: YYYY-MM-DDTHH:MM:SSZ
        let s = secs % 60;
        let m = (secs / 60) % 60;
        let h = (secs / 3600) % 24;
        let days = secs / 86400;
        // Days since 1970-01-01 → Gregorian calendar
        let (year, month, day) = days_to_ymd(days);
        format!("{year:04}-{month:02}-{day:02}T{h:02}:{m:02}:{s:02}Z")
    };

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

/// Convert days since Unix epoch (1970-01-01) to (year, month, day).
///
/// Pure arithmetic — no dependencies. Simplified Gregorian calendar calculation.
fn days_to_ymd(days: u64) -> (u64, u64, u64) {
    // Gregorian calendar: 365.2425 days/year average
    // Use the proleptic Gregorian calendar algorithm
    let z = days + 719468;
    let era = z / 146097;
    let doe = z % 146097;
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    (y, m, d)
}

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

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

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

    struct TestCredentialProvider {
        key: Mutex<Option<String>>,
    }

    impl TestCredentialProvider {
        fn with_key(key: &str) -> Arc<Self> {
            Arc::new(Self {
                key: Mutex::new(Some(key.to_string())),
            })
        }

        fn empty() -> Arc<Self> {
            Arc::new(Self {
                key: Mutex::new(None),
            })
        }

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

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

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

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

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

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

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

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

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

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

        // Close channel
        drop(tx);

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        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(),
            ..Default::default()
        };

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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