openlatch-client 0.1.18

//! Policy bundle poller — the daemon's only writer of the resident bundle.
//!
//! Modelled on `src/daemon/config_monitor/alerts.rs::run_long_poll`: the same
//! [`CredentialProvider`] + `bearer_auth` shape, the same spawn-a-loop
//! structure, the same shared `reqwest::Client`.
//!
//! # What this loop is actually defending against
//!
//! Everything here is local-authoritative. The verdict path never calls the
//! platform, so this poller is the *only* thing that can change what the host
//! enforces — and every failure mode below is therefore a question of "what
//! happens to the rules already running", not "what does the user see now".
//!
//! Three terms, used precisely and never interchanged:
//!
//! - **fail-open** — allow everything. Reserved for the two cases where there
//!   is genuinely no policy to apply: no bundle has ever been fetched, and the
//!   hook binary being unable to reach the daemon at all.
//! - **fail-static** / **last-known-good** — a refresh failed and the previously
//!   activated bundle keeps enforcing. This is what *every* branch below does.
//! - **reject** — a response was received and refused. The last-known-good
//!   bundle stays active and the stored validator is left untouched so the next
//!   poll re-attempts the same bundle.
//!
//! # The four invariants that are easy to break and expensive to debug
//!
//! 1. **D25 — store the ETag of the last successful *activation*, never of the
//!    last *reception*.** `bundle.meta.json.etag` is written only after the
//!    digest, `organization_id`, `schema_version` and `signature` all verify and
//!    the hot-swap has completed. Recording a received-but-rejected tag is the
//!    bug OPA shipped (opa#2220 → PR #2286): the next poll `304`s, and the
//!    client is wedged on the old bundle forever with no path back.
//! 2. **D26 — a `304` never clears the stored validator**, with or without an
//!    `ETag` header. RFC 9111 §4.3.4: fields absent from a `304` leave stored
//!    values *unchanged*; omission never deletes. Some origins and proxies omit
//!    it (opa#1928, opa#3361 → PR #3362, reproduced against nginx and Azure Blob
//!    Storage). Clearing it makes every other poll a full download — a total,
//!    silent degradation of the ETag transport. This branch is deliberately
//!    **separate** from the transport-error branch; do not merge them.
//! 3. **Always verify the digest against the received bytes.** Never
//!    re-serialize before hashing, and never treat a `304` as proof of
//!    integrity. A weak (`W/`) tag from a compressing proxy is accepted as a
//!    *carrier* of the expected digest, but the digest is still recomputed over
//!    the body.
//! 4. **D46 — this loop reads [`CloudState::is_auth_error`] and never writes
//!    it.** The cloud worker's credential-poll loop is the sole owner of that
//!    latch and clears it on rotation. Two writers to one latch is how you get a
//!    flag that never clears.

use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use secrecy::ExposeSecret;

use crate::cloud::{CloudState, CredentialProvider};
use crate::config::PolicyConfig;
use crate::core::error::{
    ERR_BUNDLE_FETCH_FAILED, ERR_BUNDLE_INVALID, ERR_BUNDLE_REJECTED, ERR_BUNDLE_STALE,
};
use crate::core::policy::{store, PolicyHandle, ResidentBundle};
use crate::generated::types::PolicyBundle;

/// The only `schema_version` this client understands.
///
/// A bundle declaring anything else is **rejected** and the previous one keeps
/// enforcing (`OL-1212`). Forward compatibility is the platform's job: it must
/// keep serving a v1 bundle to v1 clients rather than expecting them to guess.
pub const SUPPORTED_SCHEMA_VERSION: i64 = 1;

/// Path of the bundle endpoint, appended to the configured API base URL.
const BUNDLE_ENDPOINT: &str = "/api/v1/policy/bundle";

/// Request header carrying this install's `agent_id` on every bundle poll.
///
/// The platform reads it to compose `client_config.agent_context` — the one
/// object about THIS agent that scoped rules (`conditions`) are evaluated
/// against, locally, on the resident bundle. The header is **omitted** when the
/// daemon has no `agent_id` yet (pre-provisioning): the platform then serves the
/// org bundle without a context, and every scoped rule matches nothing — the
/// designed degrade, not an error.
///
/// The value is `config.agent_id` — the same string the alerts long-poll sends
/// as `X-OpenLatch-Machine-Id` (`config_monitor/alerts.rs`, D-2.07), and both
/// are spawned from the same `config.agent_id.clone()` in `daemon/mod.rs`. Two
/// header names for one id is deliberate: that endpoint is machine-scoped
/// (which host is asking for its alerts), this one is agent-scoped (which agent
/// row the context belongs to), and renaming either is a platform-side contract
/// change.
const AGENT_ID_HEADER: &str = "X-OpenLatch-Agent-Id";

/// Upper bound applied to a server-supplied `Retry-After`.
///
/// An origin (or something impersonating one on a hostile network) must not be
/// able to park the poller for a week with one header. The resident bundle
/// keeps enforcing throughout, so the cap only bounds how stale it may get.
const MAX_RETRY_AFTER_SECS: u64 = 3_600;

/// Floor applied to the configured poll interval.
///
/// `poll_interval_secs = 0` — reachable from `config.toml` or from
/// `OPENLATCH_POLICY_POLL_INTERVAL_SECS=0` — makes [`jittered`] return
/// `Duration::ZERO`, and the loop below then becomes a tight request loop
/// against the bundle endpoint. That is the one thing the PRD's 429 handling
/// says must never happen, and a single mistyped env var would otherwise turn
/// every host in a fleet into a hammer.
///
/// Clamping rather than refusing to start is deliberate: a daemon that fails to
/// boot on a bad interval stops enforcing policy entirely, which is a strictly
/// worse outcome than polling less often than someone asked.
const MIN_POLL_INTERVAL_SECS: u64 = 30;

// ---------------------------------------------------------------------------
// Outcome
// ---------------------------------------------------------------------------

/// What one poll attempt did. Returned so the loop can schedule the next tick
/// and so the tests can assert on the state machine without reading logs.
#[derive(Debug, Clone, PartialEq, Eq)]
enum PollOutcome {
    /// A new bundle verified and is now resident.
    Activated { revision: i64 },
    /// `304` — the resident bundle is current. The poll clock advances; nothing
    /// else changes.
    NotModified,
    /// A response was received and refused. Carries the `OL-####` code it was
    /// logged under. The previous bundle is still enforcing.
    Rejected(&'static str),
    /// `404` — the organization has no bundle. Any resident bundle **keeps
    /// enforcing**; the poll clock does not advance.
    NoBundle,
    /// `401` / `403`. Fetching pauses until the credential changes.
    AuthFailed,
    /// `429`. Carries the parsed `Retry-After`, if the server sent a usable one.
    RateLimited(Option<Duration>),
    /// `5xx`, a transport error, or an unexpected status. Retry next interval.
    Failed,
    /// No request was made. Carries a stable reason string for the log.
    Skipped(&'static str),
}

// ---------------------------------------------------------------------------
// Entry point
// ---------------------------------------------------------------------------

/// Run the policy bundle poller for the daemon's lifetime.
///
/// `handle` is the [`PolicyHandle`] the verdict path reads; `last_fetch_ok` and
/// `last_poll_ok_at` are poll state that changes *without* the bundle changing
/// (which is exactly why they are not fields on [`ResidentBundle`]):
/// `handlers.rs` reads the first for `olpolicyoffline` and `/metrics` reads the
/// second for `policy_last_poll_ok_secs`.
///
/// `base_dir` is the OpenLatch data directory (`~/.openlatch`); the store
/// appends `policy/` itself.
///
/// `agent_id` is the install's persistent id from `config.toml`, sent as
/// `X-OpenLatch-Agent-Id` on every poll so the platform can compose this agent's
/// `client_config.agent_context`. It is a parameter rather than a field of
/// [`PolicyConfig`] on purpose: `[policy]` is user-settable TOML, and the id
/// is provisioned, not configured. `None` omits the header.
#[allow(clippy::too_many_arguments)]
pub async fn run_policy_poller(
    handle: PolicyHandle,
    last_fetch_ok: Arc<AtomicBool>,
    last_poll_ok_at: Arc<AtomicI64>,
    cloud_state: CloudState,
    credentials: Arc<dyn CredentialProvider>,
    api_url: String,
    policy_config: PolicyConfig,
    base_dir: PathBuf,
    http_client: reqwest::Client,
    agent_id: Option<String>,
) {
    let mut poller = PolicyPoller::new(
        handle,
        last_fetch_ok,
        last_poll_ok_at,
        cloud_state,
        credentials,
        api_url,
        policy_config,
        base_dir,
        http_client,
        agent_id,
    );
    poller.seed_poll_clock();

    // Immediate boot fetch — BEFORE the first timer tick. Without it a freshly
    // started daemon runs on whatever is on disk (or on nothing at all) for up
    // to a full `poll_interval_secs`, which on the shipped default is five
    // minutes of a host being out of date every time it restarts.
    let mut outcome = poller.poll_once().await;
    poller.check_staleness();

    loop {
        let mut delay = jittered(poller.config.poll_interval_secs.max(MIN_POLL_INTERVAL_SECS));
        // 429 — never tight-loop. Honour `Retry-After` when it asks for longer
        // than the interval we were going to wait anyway.
        if let PollOutcome::RateLimited(Some(retry_after)) = outcome {
            if retry_after > delay {
                delay = retry_after;
            }
        }
        tokio::time::sleep(delay).await;
        outcome = poller.poll_once().await;
        poller.check_staleness();
    }
}

/// 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
/// poller retries at the 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.
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)
}

// ---------------------------------------------------------------------------
// Jitter (D38)
// ---------------------------------------------------------------------------

/// ±10% on **every** interval.
///
/// A fleet of thousands on a fixed 300 s timer is a synchronized thundering
/// herd against a single Fly-hosted origin, and **no** edge-level mitigation is
/// assumed — the CDN request-collapsing claims that would have provided one were
/// refuted during research.
///
/// Entropy source: [`uuid::Uuid::new_v4`], which is CSPRNG-backed and already a
/// dependency with the `v4` feature.
///
/// - REJECTED — adding `rand` + `getrandom` for eight lines of arithmetic.
/// - REJECTED — a deterministic per-host offset hashed from `agent_id`. It
///   de-synchronises the fleet but makes the *same* host poll at a fixed phase
///   forever, which fails the acceptance criterion "at least one consecutive
///   pair of intervals differs".
fn jittered(base: u64) -> Duration {
    let b = uuid::Uuid::new_v4().as_bytes()[0] as u64; // 0..=255
    let span = base.saturating_mul(20) / 100; // ±10% => a 20% total span
    let offset = (b * span) / 255;
    Duration::from_secs(base.saturating_sub(span / 2).saturating_add(offset))
}

// ---------------------------------------------------------------------------
// ETag parsing
// ---------------------------------------------------------------------------

/// Extract the opaque tag from an `ETag` header value.
///
/// An entity-tag is `[ W/ ] DQUOTE <etagc...> DQUOTE` (RFC 9110 §8.8.3 ABNF) —
/// the quotes are **delimiters** and are excluded from `etagc`. Both the
/// optional `W/` and the surrounding quotes must come off before comparing to
/// `sha256:<hex>`.
///
/// Stripping only the `W/` leaves the value quoted, so the comparison never
/// matches and a weak tag can never activate. Weak tags are accepted because we
/// did not mint them: nginx ≥ 1.7.3 and Cloudflare downgrade strong tags when
/// they compress. The tag is **never** proof of integrity — it is only the
/// carrier of the expected digest, which is always recomputed over the received
/// bytes.
fn parse_etag(raw: &str) -> Option<&str> {
    let t = raw.trim();
    let t = t.strip_prefix("W/").unwrap_or(t).trim();
    t.strip_prefix('"')?.strip_suffix('"')
}

// ---------------------------------------------------------------------------
// Poller
// ---------------------------------------------------------------------------

/// One poll loop's mutable state, split out from [`run_policy_poller`] so a
/// single tick is directly testable without waiting on a timer.
struct PolicyPoller {
    handle: PolicyHandle,
    last_fetch_ok: Arc<AtomicBool>,
    last_poll_ok_at: Arc<AtomicI64>,
    cloud_state: CloudState,
    credentials: Arc<dyn CredentialProvider>,
    url: String,
    config: PolicyConfig,
    base_dir: PathBuf,
    http: reqwest::Client,
    /// This install's `agent_id`, sent as [`AGENT_ID_HEADER`]. `None` before
    /// provisioning — the header is then omitted, never sent empty.
    agent_id: Option<String>,
    /// The credential bytes that last received a `401`/`403`.
    ///
    /// "Stop polling" must **not** mean "terminate the loop". This poller is
    /// forbidden from *writing* [`CloudState::is_auth_error`] (D46), so if it
    /// simply exited, nothing could ever restart it — and if the event stream
    /// happened to be idle, the cloud worker would never `401` either, never
    /// latch, and never clear. Policy updates would be permanently dead on that
    /// host with no signal anywhere.
    ///
    /// Instead: remember the credential that failed, keep looping on the normal
    /// jittered interval, and skip the request while the current credential is
    /// byte-identical to the failed one. The worker's existing 60 s
    /// credential-poll rotates it; the next tick after that fetches again.
    failed_credential: Option<Vec<u8>>,
}

impl PolicyPoller {
    #[allow(clippy::too_many_arguments)]
    fn new(
        handle: PolicyHandle,
        last_fetch_ok: Arc<AtomicBool>,
        last_poll_ok_at: Arc<AtomicI64>,
        cloud_state: CloudState,
        credentials: Arc<dyn CredentialProvider>,
        api_url: String,
        config: PolicyConfig,
        base_dir: PathBuf,
        http: reqwest::Client,
        agent_id: Option<String>,
    ) -> Self {
        // `agent_id` reaches here from `config.toml`, so it can be anything a
        // hand-edit puts there. `reqwest` would reject a value carrying a
        // newline or a non-visible-ASCII byte at request-build time on EVERY
        // tick, and the poll would fail with a transport error that names the
        // header rather than the file. Drop it once instead, with a line that
        // says what was lost: the header is then omitted exactly as it is
        // pre-provisioning, and the daemon keeps polling.
        let agent_id = agent_id.filter(|id| {
            reqwest::header::HeaderValue::from_str(id)
                .inspect_err(|_| {
                    tracing::warn!(
                        target: "policy",
                        agent_id = ?id,
                        "configured agent_id is not a usable HTTP header value; polling without it. The platform serves the org bundle with no agent_context, so every scoped rule matches nothing"
                    );
                })
                .is_ok()
        });
        Self {
            handle,
            last_fetch_ok,
            last_poll_ok_at,
            cloud_state,
            credentials,
            url: format!("{}{}", api_url.trim_end_matches('/'), BUNDLE_ENDPOINT),
            config,
            base_dir,
            http,
            agent_id,
            failed_credential: None,
        }
    }

    /// Seed the in-memory staleness clock from `bundle.meta.json`.
    ///
    /// The `OL-1213` clock measures **connectivity**, and connectivity does not
    /// reset because the daemon restarted. Without this, every restart would
    /// silence a staleness warning that has been true for days.
    fn seed_poll_clock(&self) {
        let Ok(Some(meta)) = store::read_meta(&self.base_dir) else {
            return;
        };
        if let Some(secs) = meta.last_poll_ok_at.as_deref().and_then(parse_unix_secs) {
            self.last_poll_ok_at.store(secs, Ordering::Relaxed);
        }
        self.last_fetch_ok
            .store(meta.last_fetch_ok, Ordering::Relaxed);
    }

    /// One poll attempt.
    async fn poll_once(&mut self) -> PollOutcome {
        // D46 — READ-ONLY. The cloud worker owns this latch and clears it on
        // credential rotation. Never set it, never clear it.
        if self.cloud_state.is_auth_error() {
            tracing::debug!(
                target: "policy",
                "cloud auth error is latched; skipping the policy poll (the resident bundle keeps enforcing)"
            );
            return PollOutcome::Skipped("auth_error_latched");
        }

        let Some(token) = self.credentials.retrieve() else {
            tracing::debug!(
                target: "policy",
                "no credential available; skipping the policy poll"
            );
            return PollOutcome::Skipped("no_credential");
        };
        let credential = token.expose_secret().as_bytes().to_vec();

        if self.failed_credential.as_deref() == Some(credential.as_slice()) {
            tracing::debug!(
                target: "policy",
                "credential unchanged since the last 401/403; skipping the policy poll"
            );
            return PollOutcome::Skipped("credential_unchanged_after_auth_failure");
        }

        // The validator we send is the tag of the last SUCCESSFUL ACTIVATION
        // (D25), read from disk rather than cached in memory precisely so there
        // is only one place it can be written.
        let meta = match store::read_meta(&self.base_dir) {
            Ok(meta) => meta,
            Err(e) => {
                // Unreadable meta is not fatal: fetch without a validator and
                // let the full download re-establish the pair.
                tracing::warn!(
                    target: "policy",
                    code = ERR_BUNDLE_FETCH_FAILED,
                    error = %e,
                    "could not read bundle.meta.json; polling without an If-None-Match validator"
                );
                None
            }
        };

        let mut req = self.http.get(&self.url).bearer_auth(token.expose_secret());
        // The validator is only valid for the identity it was fetched under:
        // the platform composes `client_config.agent_context` from the header
        // below, so the same URL serves a different body per agent. Revalidating
        // with a tag minted under a different (or no) identity invites a `304`
        // for a body this install has never seen — a fleet upgrading onto this
        // build would keep a context-less bundle, and every scoped rule would
        // match nothing, until the org's rules next changed. Absent-to-present
        // counts as a change, which is why this compares `Option`s.
        let validator = meta
            .as_ref()
            .filter(|m| m.agent_id.as_deref() == self.agent_id.as_deref())
            .and_then(|m| m.etag.as_deref());
        if let Some(etag) = validator {
            req = req.header(reqwest::header::IF_NONE_MATCH, etag);
        }
        // Who is asking — so the platform can serve THIS agent's context
        // alongside the org rules. Omitted (not sent empty) when unprovisioned.
        if let Some(agent_id) = self.agent_id.as_deref() {
            req = req.header(AGENT_ID_HEADER, agent_id);
        }

        let resp = match req.send().await {
            Ok(resp) => resp,
            Err(e) => {
                return self.transport_failure(&format!("policy bundle request failed: {e}"));
            }
        };

        let status = resp.status();
        // Any response at all proves the credential is being evaluated again,
        // so a stale pause must not survive it.
        if status.as_u16() != 401 && status.as_u16() != 403 {
            self.failed_credential = None;
        }

        match status.as_u16() {
            200 => self.handle_ok(resp, meta).await,
            304 => self.handle_not_modified(meta),
            401 | 403 => {
                self.failed_credential = Some(credential);
                tracing::error!(
                    target: "policy",
                    code = ERR_BUNDLE_FETCH_FAILED,
                    status = status.as_u16(),
                    "policy bundle fetch rejected the credential; pausing bundle refresh until the credential changes. The resident bundle keeps enforcing"
                );
                self.record_poll_failure(&format!(
                    "{ERR_BUNDLE_FETCH_FAILED} auth rejected ({})",
                    status.as_u16()
                ));
                PollOutcome::AuthFailed
            }
            404 => {
                // A transient 404 must NEVER silently disarm a host: removing
                // policy is done by disabling rules or by the kill switch, never
                // by the bundle endpoint going missing for a minute.
                //
                // The poll clock deliberately does NOT advance. A *permanent*
                // 404 — the server genuinely lost this org's bundle — would
                // otherwise suppress the OL-1213 staleness warning forever.
                let resident = self.handle.load().is_some();
                tracing::warn!(
                    target: "policy",
                    code = ERR_BUNDLE_FETCH_FAILED,
                    resident_bundle = resident,
                    "policy bundle endpoint returned 404; keeping the last-known-good bundle enforcing"
                );
                self.record_poll_failure(&format!("{ERR_BUNDLE_FETCH_FAILED} 404 no bundle"));
                PollOutcome::NoBundle
            }
            429 => {
                let retry_after = resp
                    .headers()
                    .get(reqwest::header::RETRY_AFTER)
                    .and_then(|v| v.to_str().ok())
                    .and_then(parse_retry_after)
                    .map(|s| Duration::from_secs(s.min(MAX_RETRY_AFTER_SECS)));
                tracing::warn!(
                    target: "policy",
                    code = ERR_BUNDLE_FETCH_FAILED,
                    retry_after_secs = retry_after.map(|d| d.as_secs()),
                    "policy bundle fetch rate limited (429); backing off"
                );
                self.record_poll_failure(&format!("{ERR_BUNDLE_FETCH_FAILED} rate limited"));
                PollOutcome::RateLimited(retry_after)
            }
            code => self.transport_failure(&format!(
                "policy bundle fetch returned an unusable status {code}"
            )),
        }
    }

    /// `200` — the only path that may activate a bundle.
    ///
    /// Verification order is load-bearing and every step keeps the previous
    /// bundle on failure:
    ///
    /// 1. an `ETag` header is present and parseable — the digest is **not**
    ///    inside the body, so an unverifiable body must never activate;
    /// 2. `sha256(received_bytes)` equals the tag's opaque value;
    /// 3. the body is JSON;
    /// 4. `schema_version` is [`SUPPORTED_SCHEMA_VERSION`];
    /// 5. the body deserialises into a [`PolicyBundle`];
    /// 6. `organization_id` matches the trust-on-first-use anchor;
    /// 7. `signature` is `null` (D32 — a client that cannot verify a signature
    ///    must reject rather than ignore it, so enabling signing later cannot be
    ///    silently downgraded by an old client).
    ///
    /// Only then: write the body, write the meta — **and only now** is the ETag
    /// persisted — and hot-swap.
    async fn handle_ok(
        &mut self,
        resp: reqwest::Response,
        meta: Option<store::BundleMeta>,
    ) -> PollOutcome {
        let raw_etag = resp
            .headers()
            .get(reqwest::header::ETAG)
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());

        let body = match resp.bytes().await {
            Ok(b) => b,
            Err(e) => {
                return self.transport_failure(&format!("policy bundle body was not readable: {e}"))
            }
        };

        let Some(raw_etag) = raw_etag else {
            return self.reject(
                ERR_BUNDLE_REJECTED,
                "200 response carried no ETag header, so the body's digest cannot be verified",
            );
        };
        let Some(tag) = parse_etag(&raw_etag) else {
            return self.reject(
                ERR_BUNDLE_REJECTED,
                "ETag header is not a well-formed entity-tag",
            );
        };
        if let Err(e) = store::verify_digest(&body, tag) {
            return self.reject(ERR_BUNDLE_REJECTED, &e.to_string());
        }

        // Peek at `schema_version` through an untyped Value first. `PolicyBundle`
        // is `deny_unknown_fields`, so a v2 document would otherwise fail
        // deserialization and be reported as "malformed JSON" — the same code,
        // but a log line that sends the reader hunting for corruption instead of
        // for a version skew.
        let value: serde_json::Value = match serde_json::from_slice(&body) {
            Ok(v) => v,
            Err(e) => {
                return self.reject(ERR_BUNDLE_INVALID, &format!("body is not valid JSON: {e}"))
            }
        };
        match value.get("schema_version").and_then(|v| v.as_i64()) {
            Some(SUPPORTED_SCHEMA_VERSION) => {}
            other => {
                return self.reject(
                    ERR_BUNDLE_INVALID,
                    &format!("unsupported schema_version {other:?}"),
                )
            }
        }
        // One rule this build cannot read must not cost the whole bundle — a
        // fleet mid-rollout would otherwise stop receiving policy the moment an
        // author uses a key only the newer clients know.
        let bundle: PolicyBundle = match crate::core::policy::parse_bundle_tolerant(value) {
            Ok(b) => b,
            Err(e) => {
                return self.reject(
                    ERR_BUNDLE_INVALID,
                    &format!("body is not a valid policy bundle: {e}"),
                )
            }
        };

        // Trust-on-first-use. The expected org is `bundle.meta.json`'s, written
        // on the FIRST activation and never rewritten. When there is none,
        // activate anyway and record this bundle's — the API key already scopes
        // the response server-side, so the first bundle is trusted on that
        // basis. The check exists to catch a LATER mismatch: a key reissued to a
        // different org, or a cached bundle copied between hosts.
        //
        // Never compare against `agent_id` — that is a machine id, not an org
        // id. `~/.openlatch/identity.json::org_id` is not usable either: it is
        // written only when telemetry consent is active, so it is absent on
        // every telemetry-disabled host and cannot be an enforcement input.
        // The in-memory fallback covers the one case disk cannot: the policy
        // directory is unwritable, so no meta was ever persisted, yet a bundle
        // is resident. Without it, TOFU would re-trust a *different* org on
        // every poll of such a host.
        let expected_org: Option<String> = meta
            .as_ref()
            .map(|m| m.organization_id.clone())
            .or_else(|| {
                self.handle
                    .load()
                    .as_ref()
                    .as_ref()
                    .map(|b| b.organization_id.clone())
            });
        if let Some(expected) = expected_org {
            if bundle.organization_id != expected {
                return self.reject(
                    ERR_BUNDLE_REJECTED,
                    &format!(
                        "organization_id {} does not match this host's {expected}",
                        bundle.organization_id
                    ),
                );
            }
        }

        if bundle.signature.is_some() {
            return self.reject(
                ERR_BUNDLE_INVALID,
                "bundle carries a signature this client cannot verify (D32 lands in v1.1)",
            );
        }

        // `verify_digest` above proved `digest_of(&body) == tag`, so the digest
        // IS the tag already in hand — hashing the body a second time would
        // only recompute a value this branch has established.
        let digest = tag.to_string();
        let mut new_meta = store::BundleMeta::activated(&bundle, digest.clone(), Some(raw_etag));
        // Written with the tag, never separately: the pair is what the next
        // poll checks before it dares revalidate.
        new_meta.agent_id = self.agent_id.clone();
        // Body first, meta second — the meta file is the commit marker. A disk
        // failure here is NOT fatal: the verified bundle still activates in
        // memory and keeps enforcing (PRD: "policy dir unwritable or disk full →
        // keep the in-memory bundle and keep enforcing; log OL-1210"). Because
        // the ETag never landed, the next poll simply re-downloads.
        if let Err(e) = store::store(&self.base_dir, &body, &new_meta) {
            tracing::warn!(
                target: "policy",
                code = ERR_BUNDLE_FETCH_FAILED,
                error = %e,
                "could not persist the policy bundle; activating it in memory anyway"
            );
        }

        let revision = bundle.revision;
        let resident = ResidentBundle::from_bundle(&bundle);
        let rule_count = resident.command_rules.len();
        // Logged separately so D-U14 rollout — "are clients accepting request
        // rules yet?" — is answerable from the daemon log, not just inferred.
        let request_rule_count = resident.request_rules.len();
        let enforcement_enabled = resident.enforcement_enabled;
        // The hot swap. An in-flight evaluation sees either the whole old bundle
        // or the whole new one, never a partially-mutated structure.
        self.handle.store(Arc::new(Some(resident)));

        self.last_fetch_ok.store(true, Ordering::Relaxed);
        self.last_poll_ok_at
            .store(now_unix_secs(), Ordering::Relaxed);

        tracing::info!(
            target: "policy",
            revision,
            digest = %digest,
            rules = rule_count,
            request_rules = request_rule_count,
            enforcement_enabled,
            "policy bundle activated"
        );
        PollOutcome::Activated { revision }
    }

    /// `304` — a **success**, and the only branch that must never touch the
    /// stored validator.
    ///
    /// The body is not re-read, the rule set is not reloaded, `bundle.json` is
    /// not rewritten, and `etag` is carried through untouched **whether or not
    /// the 304 carried an ETag header** (D26). Kept deliberately separate from
    /// [`Self::transport_failure`].
    fn handle_not_modified(&self, meta: Option<store::BundleMeta>) -> PollOutcome {
        self.last_fetch_ok.store(true, Ordering::Relaxed);
        self.last_poll_ok_at
            .store(now_unix_secs(), Ordering::Relaxed);

        if let Some(mut meta) = meta {
            // Read-modify-write: `etag`, `digest`, `revision` and
            // `organization_id` all survive by construction. Only the poll tier
            // moves — the download and activation tiers stay where they were,
            // which is the whole point of splitting them (D28).
            meta.last_poll_ok_at = Some(store::now_rfc3339());
            meta.last_fetch_ok = true;
            meta.last_error = None;
            if let Err(e) = store::write_meta(&self.base_dir, &meta) {
                tracing::warn!(
                    target: "policy",
                    code = ERR_BUNDLE_FETCH_FAILED,
                    error = %e,
                    "could not update the policy poll clock on disk"
                );
            }
        }

        tracing::debug!(
            target: "policy",
            "policy bundle unchanged (304); the resident bundle stays active"
        );
        PollOutcome::NotModified
    }

    /// A response was received and refused. The previous bundle keeps enforcing
    /// and the stored validator is untouched, so the next poll re-requests the
    /// same bundle and re-attempts activation (D25).
    fn reject(&self, code: &'static str, detail: &str) -> PollOutcome {
        tracing::warn!(
            target: "policy",
            code,
            detail,
            "policy bundle rejected; keeping the last-known-good bundle enforcing"
        );
        self.record_poll_failure(&format!("{code} {detail}"));
        PollOutcome::Rejected(code)
    }

    /// 5xx, a transport error, or an unusable status. Fail-static: retry next
    /// interval.
    fn transport_failure(&self, detail: &str) -> PollOutcome {
        tracing::warn!(
            target: "policy",
            code = ERR_BUNDLE_FETCH_FAILED,
            detail,
            "policy bundle poll failed; keeping the last-known-good bundle enforcing"
        );
        self.record_poll_failure(&format!("{ERR_BUNDLE_FETCH_FAILED} {detail}"));
        PollOutcome::Failed
    }

    /// Mark the poll as unsuccessful without disturbing anything the next poll
    /// depends on.
    ///
    /// `last_poll_ok_at` does **not** advance (that is the `OL-1213` clock) and
    /// `etag` is preserved by read-modify-write. A disk error here is logged at
    /// `debug` and swallowed: the in-memory bundle is what enforces, and failing
    /// to write a diagnostic must never become a second failure to handle.
    fn record_poll_failure(&self, message: &str) {
        self.last_fetch_ok.store(false, Ordering::Relaxed);
        let Ok(Some(mut meta)) = store::read_meta(&self.base_dir) else {
            return;
        };
        meta.last_fetch_ok = false;
        meta.last_error = Some(message.to_string());
        if let Err(e) = store::write_meta(&self.base_dir, &meta) {
            tracing::debug!(
                target: "policy",
                error = %e,
                "could not record the policy poll failure on disk"
            );
        }
    }

    /// Warn (`OL-1213`) when no poll has succeeded within
    /// `stale_warn_after_secs` — **and keep enforcing**. Returns whether the
    /// warning fired, for the tests.
    ///
    /// Measured from the last successful **poll** (connectivity), never from the
    /// bundle's `built_at` (policy age): a `304` counts, so a healthy
    /// organization whose rules simply never change does not warn forever.
    ///
    /// A clock of `0` means no poll has ever succeeded on this host, which is
    /// not measurable staleness — a host that has never reached the platform
    /// either has no bundle (already reported as such) or has a disk bundle
    /// whose freshness nothing here can speak to.
    fn check_staleness(&self) -> bool {
        let last = self.last_poll_ok_at.load(Ordering::Relaxed);
        if last <= 0 {
            return false;
        }
        let age = now_unix_secs().saturating_sub(last).max(0) as u64;
        if age <= self.config.stale_warn_after_secs {
            return false;
        }
        tracing::warn!(
            target: "policy",
            code = ERR_BUNDLE_STALE,
            stale_seconds = age,
            threshold_seconds = self.config.stale_warn_after_secs,
            "no successful policy bundle poll within the staleness threshold; the resident bundle keeps enforcing"
        );
        true
    }
}

// ---------------------------------------------------------------------------
// Time helpers
// ---------------------------------------------------------------------------

/// Now, as Unix seconds. `0` on a clock before the epoch, which is the same
/// sentinel the atomics use for "never".
fn now_unix_secs() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

/// Parse an RFC 3339 timestamp into Unix seconds.
fn parse_unix_secs(raw: &str) -> Option<i64> {
    chrono::DateTime::parse_from_rfc3339(raw)
        .ok()
        .map(|dt| dt.timestamp())
}

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

    use std::sync::atomic::AtomicUsize;
    use std::sync::Mutex;

    use secrecy::SecretString;

    use crate::core::policy::test_support::wire_rule;
    use crate::core::policy::{evaluate_command, new_handle};
    use crate::generated::types::{PolicyRule, PolicyRuleMode, PolicyRuleSeverity};

    const ORG: &str = "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42";
    const OTHER_ORG: &str = "0192f8a1-0000-0000-0000-000000000000";

    // -----------------------------------------------------------------------
    // Fixtures
    // -----------------------------------------------------------------------

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

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

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

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

    /// Thin wrapper over the shared fixture, so the wire shape is written once.
    /// A second hand-written `PolicyRule` literal here has to be patched in
    /// lockstep every time the generated type gains a field — which is exactly
    /// what adding the request plane would otherwise have required.
    fn rule(rule_id: &str, pattern: &str, mode: PolicyRuleMode) -> PolicyRule {
        wire_rule(rule_id, pattern, mode, PolicyRuleSeverity::High)
    }

    fn bundle_json(revision: i64, org: &str, rules: Vec<PolicyRule>) -> serde_json::Value {
        serde_json::json!({
            "schema_version": 1,
            "organization_id": org,
            "revision": revision,
            "built_at": "2026-07-21T09:00:00Z",
            "enforcement_enabled": true,
            "rules": rules,
            "signature": serde_json::Value::Null,
        })
    }

    /// The canonical fixture: one enforcing rule that denies `rm -rf /tmp`.
    fn body(revision: i64) -> Vec<u8> {
        serde_json::to_vec(&bundle_json(
            revision,
            ORG,
            vec![rule("OL-CMD-001", "*rm -rf*", PolicyRuleMode::Enforce)],
        ))
        .expect("serialise fixture")
    }

    fn etag_for(body: &[u8]) -> String {
        format!("\"{}\"", store::digest_of(body))
    }

    struct Harness {
        poller: PolicyPoller,
        dir: tempfile::TempDir,
        credentials: Arc<TestCredentialProvider>,
        cloud_state: CloudState,
        handle: PolicyHandle,
        last_fetch_ok: Arc<AtomicBool>,
        last_poll_ok_at: Arc<AtomicI64>,
    }

    /// The `agent_id` every provisioned-install test polls with.
    const AGENT_ID: &str = "0192f8a1-4c3b-7e2a-9f10-5d8c3b1a7e42";

    impl Harness {
        fn new(api_url: String) -> Self {
            Self::with_agent_id(api_url, Some(AGENT_ID.to_string()))
        }

        /// The provisioning state is the one knob the harness exposes: `None`
        /// is a daemon whose `config.toml` has no `agent_id` yet.
        fn with_agent_id(api_url: String, agent_id: Option<String>) -> Self {
            let dir = tempfile::tempdir().expect("tempdir");
            let handle = new_handle(None);
            let last_fetch_ok = Arc::new(AtomicBool::new(false));
            let last_poll_ok_at = Arc::new(AtomicI64::new(0));
            let cloud_state = CloudState::new();
            let credentials = TestCredentialProvider::with_key("ol_org_test");
            let poller = PolicyPoller::new(
                handle.clone(),
                last_fetch_ok.clone(),
                last_poll_ok_at.clone(),
                cloud_state.clone(),
                credentials.clone(),
                api_url,
                PolicyConfig {
                    enabled: true,
                    poll_interval_secs: 300,
                    stale_warn_after_secs: 86_400,
                },
                dir.path().to_path_buf(),
                reqwest::Client::new(),
                agent_id,
            );
            Self {
                poller,
                dir,
                credentials,
                cloud_state,
                handle,
                last_fetch_ok,
                last_poll_ok_at,
            }
        }

        /// Is the resident bundle still denying the canary command?
        fn still_enforcing(&self) -> bool {
            let loaded = self.handle.load();
            let Some(bundle) = loaded.as_ref().as_ref() else {
                return false;
            };
            evaluate_command(bundle, "rm -rf /tmp").is_some_and(|m| !m.shadow)
        }

        fn resident_revision(&self) -> Option<i64> {
            self.handle.load().as_ref().as_ref().map(|b| b.revision)
        }

        fn meta(&self) -> Option<store::BundleMeta> {
            store::read_meta(self.dir.path()).expect("meta readable")
        }
    }

    // -----------------------------------------------------------------------
    // Pure helpers
    // -----------------------------------------------------------------------

    #[test]
    fn parse_etag_strips_quotes_and_the_weak_prefix() {
        assert_eq!(parse_etag("\"sha256:abc\""), Some("sha256:abc"));
        assert_eq!(parse_etag("W/\"sha256:abc\""), Some("sha256:abc"));
        assert_eq!(parse_etag("  W/ \"sha256:abc\" "), Some("sha256:abc"));
        // Stripping only the `W/` would leave a quoted value that can never
        // equal `sha256:<hex>` — the bug this function exists to not have.
        assert_ne!(parse_etag("W/\"sha256:abc\""), Some("\"sha256:abc\""));
        // Not an entity-tag at all.
        assert_eq!(parse_etag("sha256:abc"), None);
        assert_eq!(parse_etag(""), None);
    }

    /// D38 — every interval lands inside ±10% of the base, and the sequence is
    /// not constant (a fixed per-host phase would de-synchronise the fleet but
    /// still fail this).
    #[test]
    fn jitter_stays_within_ten_percent_and_varies() {
        let base = 300u64;
        let intervals: Vec<u64> = (0..20).map(|_| jittered(base).as_secs()).collect();
        for secs in &intervals {
            assert!(
                (270..=330).contains(secs),
                "interval {secs}s outside ±10% of {base}s"
            );
        }
        assert!(
            intervals.windows(2).any(|w| w[0] != w[1]),
            "at least one consecutive pair must differ: {intervals:?}"
        );
    }

    #[test]
    fn jitter_does_not_underflow_on_a_tiny_interval() {
        for _ in 0..20 {
            let _ = jittered(1);
            let _ = jittered(0);
        }
    }

    #[test]
    fn a_zero_poll_interval_cannot_produce_a_tight_loop() {
        // `poll_interval_secs = 0` is reachable from config.toml and from
        // OPENLATCH_POLICY_POLL_INTERVAL_SECS. Unclamped, `jittered(0)` is
        // Duration::ZERO and the poll loop becomes a tight request loop against
        // the bundle endpoint — every host in the fleet turning into a hammer
        // over one mistyped env var.
        // Mirrors what config.toml / OPENLATCH_POLICY_POLL_INTERVAL_SECS supply.
        let configured: u64 = 0;
        assert_eq!(
            jittered(configured),
            Duration::ZERO,
            "precondition: 0 really is degenerate"
        );
        for _ in 0..50 {
            let delay = jittered(configured.max(MIN_POLL_INTERVAL_SECS));
            assert!(
                delay >= Duration::from_secs(MIN_POLL_INTERVAL_SECS * 9 / 10),
                "clamped interval fell below the floor: {delay:?}"
            );
        }
    }

    // -----------------------------------------------------------------------
    // Retry-After (RFC 9110 §10.2.3 — BOTH forms)
    // -----------------------------------------------------------------------

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

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

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

    // -----------------------------------------------------------------------
    // 200 — activation
    // -----------------------------------------------------------------------

    #[tokio::test]
    async fn ok_activates_and_stores_the_activation_etag() {
        let mut server = mockito::Server::new_async().await;
        let body = body(42);
        let etag = etag_for(&body);
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag)
            .with_body(&body)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Activated { revision: 42 }
        );

        assert!(h.still_enforcing(), "the new bundle must be enforcing");
        assert_eq!(h.resident_revision(), Some(42));
        let meta = h.meta().expect("meta written");
        assert_eq!(meta.etag.as_deref(), Some(etag.as_str()));
        assert_eq!(meta.digest, store::digest_of(&body));
        assert_eq!(meta.organization_id, ORG);
        assert!(meta.last_activated_at.is_some());
        assert!(h.last_fetch_ok.load(Ordering::Relaxed));
        assert!(h.last_poll_ok_at.load(Ordering::Relaxed) > 0);
        // The bytes on disk are the bytes received — never re-serialized.
        assert_eq!(
            std::fs::read(store::bundle_path(h.dir.path())).expect("body on disk"),
            body
        );
        mock.assert_async().await;
    }

    /// A weak tag from a compressing proxy (nginx ≥ 1.7.3, Cloudflare) still
    /// carries a usable digest.
    #[tokio::test]
    async fn weak_etag_is_accepted() {
        let mut server = mockito::Server::new_async().await;
        let body = body(7);
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &format!("W/\"{}\"", store::digest_of(&body)))
            .with_body(&body)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Activated { revision: 7 }
        );
        assert!(h.still_enforcing());
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn empty_rule_set_activates_and_keeps_a_revision() {
        let mut server = mockito::Server::new_async().await;
        let body = serde_json::to_vec(&bundle_json(9, ORG, vec![])).expect("serialise");
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag_for(&body))
            .with_body(&body)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Activated { revision: 9 }
        );
        // "Allow everything, revision 9" — telemetry can still distinguish this
        // from "no bundle", which is the whole point (D29).
        assert!(!h.still_enforcing());
        assert_eq!(h.resident_revision(), Some(9));
        mock.assert_async().await;
    }

    // -----------------------------------------------------------------------
    // X-OpenLatch-Agent-Id (C-01) — who is asking
    // -----------------------------------------------------------------------

    /// A provisioned install identifies itself on every poll, so the platform
    /// can compose `client_config.agent_context` for THIS agent. mockito's
    /// header match is exact: a missing or misspelled header falls through to
    /// its 501 default and the poll would read as `Failed`, not `Activated`.
    #[tokio::test]
    async fn a_provisioned_install_sends_its_agent_id_on_every_poll() {
        let mut server = mockito::Server::new_async().await;
        let body = body(11);
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("x-openlatch-agent-id", AGENT_ID)
            .with_status(200)
            .with_header("ETag", &etag_for(&body))
            .with_body(&body)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Activated { revision: 11 }
        );
        mock.assert_async().await;
    }

    /// The header rides beside `If-None-Match` — identity does not displace
    /// the validator, and a `304` for a known agent is still a `304`.
    #[tokio::test]
    async fn the_agent_id_header_rides_beside_the_validator() {
        let mut server = mockito::Server::new_async().await;
        let body = body(12);
        let etag = etag_for(&body);
        let first = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("x-openlatch-agent-id", AGENT_ID)
            .with_status(200)
            .with_header("ETag", &etag)
            .with_body(&body)
            .expect(1)
            .create_async()
            .await;
        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        first.assert_async().await;

        let revalidate = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("if-none-match", etag.as_str())
            .match_header("x-openlatch-agent-id", AGENT_ID)
            .with_status(304)
            .expect(1)
            .create_async()
            .await;
        assert_eq!(h.poller.poll_once().await, PollOutcome::NotModified);
        revalidate.assert_async().await;
    }

    /// Pre-provisioning (no `agent_id` in `config.toml`) the header is
    /// **omitted** — never sent empty. The platform serves the org bundle
    /// without a context, and every scoped rule matches nothing: the designed
    /// degrade, so the poll itself must still succeed.
    #[tokio::test]
    async fn an_unprovisioned_install_omits_the_agent_id_header() {
        let mut server = mockito::Server::new_async().await;
        let body = body(13);
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("x-openlatch-agent-id", mockito::Matcher::Missing)
            .with_status(200)
            .with_header("ETag", &etag_for(&body))
            .with_body(&body)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::with_agent_id(server.url(), None);
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Activated { revision: 13 }
        );
        mock.assert_async().await;
    }

    /// A cached validator belongs to the identity that obtained it. A fleet
    /// upgrading onto this build has meta files written by a pre-header client:
    /// their `etag` is for the org bundle served with no `agent_context`, so
    /// echoing it would earn a `304` and leave the install context-less —
    /// scoped rules matching nothing — until the org's rules next changed.
    /// The first poll after an upgrade must therefore be a full download.
    #[tokio::test]
    async fn a_validator_fetched_under_another_identity_is_not_revalidated() {
        let mut server = mockito::Server::new_async().await;
        let body = body(14);
        let etag = etag_for(&body);

        // Seed the disk the way a pre-header build left it: a perfectly good
        // body + validator, and no `agent_id` key in the file at all.
        let mut h = Harness::new(server.url());
        let mut meta = store::BundleMeta::activated(
            &serde_json::from_slice(&body).expect("fixture parses"),
            store::digest_of(&body),
            Some(etag.clone()),
        );
        meta.agent_id = None;
        let mut raw = serde_json::to_value(&meta).expect("meta serialises");
        raw.as_object_mut().expect("object").remove("agent_id");
        store::write_body(h.dir.path(), &body).expect("write body");
        std::fs::write(
            store::meta_path(h.dir.path()),
            serde_json::to_vec(&raw).expect("serialise"),
        )
        .expect("write legacy meta");
        assert!(
            h.meta().expect("meta readable").agent_id.is_none(),
            "precondition: the legacy file carries no identity"
        );

        let full = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("if-none-match", mockito::Matcher::Missing)
            .match_header("x-openlatch-agent-id", AGENT_ID)
            .with_status(200)
            .with_header("ETag", &etag)
            .with_body(&body)
            .expect(1)
            .create_async()
            .await;
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Activated { revision: 14 }
        );
        full.assert_async().await;

        // The activation just bound the tag to this poller's identity, so the
        // next poll is a cheap revalidation again.
        assert_eq!(h.meta().expect("meta").agent_id.as_deref(), Some(AGENT_ID));
        let revalidate = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("if-none-match", etag.as_str())
            .match_header("x-openlatch-agent-id", AGENT_ID)
            .with_status(304)
            .expect(1)
            .create_async()
            .await;
        assert_eq!(h.poller.poll_once().await, PollOutcome::NotModified);
        revalidate.assert_async().await;
    }

    /// The same binding in the other direction: an install that had an
    /// `agent_id` and lost it (re-provisioned, or the key removed from
    /// `config.toml`) must not reuse the context-bearing validator either.
    #[tokio::test]
    async fn losing_the_agent_id_also_drops_the_validator() {
        let mut server = mockito::Server::new_async().await;
        let body = body(15);
        let etag = etag_for(&body);
        let seed = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag)
            .with_body(&body)
            .expect(1)
            .create_async()
            .await;
        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        seed.assert_async().await;

        h.poller.agent_id = None;
        let full = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("if-none-match", mockito::Matcher::Missing)
            .match_header("x-openlatch-agent-id", mockito::Matcher::Missing)
            .with_status(200)
            .with_header("ETag", &etag)
            .with_body(&body)
            .expect(1)
            .create_async()
            .await;
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        full.assert_async().await;
    }

    /// A hand-edited `config.toml` must not wedge polling. An `agent_id` that
    /// cannot be a header value is dropped at construction — the poll still
    /// happens, without the header, exactly as it does pre-provisioning.
    #[tokio::test]
    async fn an_agent_id_that_is_not_a_header_value_is_dropped_not_sent() {
        let mut server = mockito::Server::new_async().await;
        let body = body(16);
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("x-openlatch-agent-id", mockito::Matcher::Missing)
            .with_status(200)
            .with_header("ETag", &etag_for(&body))
            .with_body(&body)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::with_agent_id(
            server.url(),
            Some(
                "bad
id"
                .to_string(),
            ),
        );
        assert!(h.poller.agent_id.is_none(), "dropped at construction");
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Activated { revision: 16 }
        );
        mock.assert_async().await;
    }

    // -----------------------------------------------------------------------
    // 200 — rejections
    // -----------------------------------------------------------------------

    async fn assert_rejected(
        builder: impl FnOnce(mockito::Mock) -> mockito::Mock,
        code: &'static str,
    ) {
        let mut server = mockito::Server::new_async().await;
        let mock = builder(server.mock("GET", BUNDLE_ENDPOINT))
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert_eq!(h.poller.poll_once().await, PollOutcome::Rejected(code));
        assert!(
            h.handle.load().is_none(),
            "a rejected bundle must never activate"
        );
        assert!(!h.last_fetch_ok.load(Ordering::Relaxed));
        assert_eq!(
            h.last_poll_ok_at.load(Ordering::Relaxed),
            0,
            "a rejection is not a successful poll"
        );
        mock.assert_async().await;
    }

    /// The digest is not inside the body, so a body with no tag is
    /// unverifiable — and an unverifiable body must never activate.
    #[tokio::test]
    async fn missing_etag_on_200_is_rejected() {
        let body = body(1);
        assert_rejected(
            move |m| m.with_status(200).with_body(body),
            ERR_BUNDLE_REJECTED,
        )
        .await;
    }

    #[tokio::test]
    async fn malformed_etag_on_200_is_rejected() {
        let body = body(1);
        assert_rejected(
            move |m| {
                m.with_status(200)
                    .with_header("ETag", "sha256:unquoted")
                    .with_body(body)
            },
            ERR_BUNDLE_REJECTED,
        )
        .await;
    }

    #[tokio::test]
    async fn digest_mismatch_is_rejected() {
        let served = body(1);
        let other = etag_for(&body(2));
        assert_rejected(
            move |m| {
                m.with_status(200)
                    .with_header("ETag", &other)
                    .with_body(served)
            },
            ERR_BUNDLE_REJECTED,
        )
        .await;
    }

    #[tokio::test]
    async fn org_mismatch_is_rejected() {
        // Seed an activation for ORG so the TOFU anchor exists…
        let mut server = mockito::Server::new_async().await;
        let first = body(1);
        let good = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag_for(&first))
            .with_body(&first)
            .expect(1)
            .create_async()
            .await;
        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        good.assert_async().await;

        // …then serve a perfectly well-formed bundle belonging to someone else.
        let intruder = serde_json::to_vec(&bundle_json(
            2,
            OTHER_ORG,
            vec![rule("OL-CMD-999", "*", PolicyRuleMode::Enforce)],
        ))
        .expect("serialise");
        let bad = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag_for(&intruder))
            .with_body(&intruder)
            .create_async()
            .await;

        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Rejected(ERR_BUNDLE_REJECTED)
        );
        assert_eq!(h.resident_revision(), Some(1), "never evaluate another org");
        bad.assert_async().await;
    }

    #[tokio::test]
    async fn unknown_schema_version_is_rejected() {
        let mut doc = bundle_json(1, ORG, vec![]);
        doc["schema_version"] = serde_json::json!(2);
        let body = serde_json::to_vec(&doc).expect("serialise");
        let etag = etag_for(&body);
        assert_rejected(
            move |m| {
                m.with_status(200)
                    .with_header("ETag", &etag)
                    .with_body(body)
            },
            ERR_BUNDLE_INVALID,
        )
        .await;
    }

    #[tokio::test]
    async fn malformed_json_is_rejected() {
        let body = b"{ not json".to_vec();
        let etag = etag_for(&body);
        assert_rejected(
            move |m| {
                m.with_status(200)
                    .with_header("ETag", &etag)
                    .with_body(body)
            },
            ERR_BUNDLE_INVALID,
        )
        .await;
    }

    /// D32 — a client that cannot verify a signature must reject rather than
    /// ignore it, so enabling signing later cannot be silently downgraded.
    #[tokio::test]
    async fn non_null_signature_is_rejected() {
        let mut doc = bundle_json(1, ORG, vec![]);
        doc["signature"] = serde_json::json!("ed25519:deadbeef");
        let body = serde_json::to_vec(&doc).expect("serialise");
        let etag = etag_for(&body);
        assert_rejected(
            move |m| {
                m.with_status(200)
                    .with_header("ETag", &etag)
                    .with_body(body)
            },
            ERR_BUNDLE_INVALID,
        )
        .await;
    }

    // -----------------------------------------------------------------------
    // D25 — the activation-ETag rewind
    // -----------------------------------------------------------------------

    /// Serve a good bundle, then a corrupted one, and prove the client is not
    /// wedged: the NEXT request still carries the PREVIOUS `If-None-Match`, and
    /// the previous bundle is still enforcing.
    ///
    /// Storing the *received* tag instead is the bug OPA shipped (opa#2220 → PR
    /// #2286): the failed bundle is recorded as activated, the next poll 304s,
    /// and there is no path back.
    #[tokio::test]
    async fn activation_failure_rewinds_the_etag() {
        let mut server = mockito::Server::new_async().await;
        let good = body(1);
        let good_etag = etag_for(&good);

        let first = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &good_etag)
            .with_body(&good)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        first.assert_async().await;

        // A corrupted revision 2: the ETag advertises a digest the body does
        // not have. It must be refused AND must not be remembered.
        let corrupt = body(2);
        let second = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("if-none-match", good_etag.as_str())
            .with_status(200)
            .with_header("ETag", &etag_for(b"a different document entirely"))
            .with_body(&corrupt)
            .expect(1)
            .create_async()
            .await;

        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Rejected(ERR_BUNDLE_REJECTED)
        );
        second.assert_async().await;

        assert_eq!(h.resident_revision(), Some(1), "revision 1 keeps enforcing");
        assert!(h.still_enforcing());
        assert_eq!(
            h.meta().expect("meta").etag.as_deref(),
            Some(good_etag.as_str()),
            "the stored validator must still be revision 1's"
        );

        // THE assertion: the third request re-attempts with the PREVIOUS
        // validator, so a fixed bundle can still be adopted.
        let fixed = body(3);
        let third = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("if-none-match", good_etag.as_str())
            .with_status(200)
            .with_header("ETag", &etag_for(&fixed))
            .with_body(&fixed)
            .expect(1)
            .create_async()
            .await;

        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Activated { revision: 3 }
        );
        third.assert_async().await;
    }

    // -----------------------------------------------------------------------
    // D26 — a 304 never clears the validator
    // -----------------------------------------------------------------------

    /// Two consecutive **bare** 304s (no `ETag` header at all — nginx and Azure
    /// Blob Storage both do this) must leave the stored validator intact, so the
    /// third request still revalidates and **zero** full downloads occur.
    ///
    /// RFC 9111 §4.3.4: fields absent from a 304 leave stored values unchanged;
    /// omission never deletes.
    #[tokio::test]
    async fn bare_304s_do_not_clear_the_validator() {
        let mut server = mockito::Server::new_async().await;
        let good = body(11);
        let etag = etag_for(&good);
        let body_on_disk = good.clone();

        let download = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag)
            .with_body(&good)
            .expect(1) // exactly one full download, ever
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        download.assert_async().await;

        // Three revalidations, all bare: no ETag header on the 304.
        let revalidate = server
            .mock("GET", BUNDLE_ENDPOINT)
            .match_header("if-none-match", etag.as_str())
            .with_status(304)
            .expect(3)
            .create_async()
            .await;

        for _ in 0..3 {
            assert_eq!(h.poller.poll_once().await, PollOutcome::NotModified);
        }
        revalidate.assert_async().await;
        // The full-download mock is still at exactly one call — asserted again
        // after the 304s because that is the regression: a cleared validator
        // turns every other poll into a download.
        download.assert_async().await;

        assert_eq!(
            h.meta().expect("meta").etag.as_deref(),
            Some(etag.as_str()),
            "a 304 must never clear the stored validator"
        );
        assert_eq!(
            std::fs::read(store::bundle_path(h.dir.path())).expect("body"),
            body_on_disk,
            "a 304 must not touch bundle.json"
        );
        assert_eq!(h.resident_revision(), Some(11));
    }

    #[tokio::test]
    async fn not_modified_advances_the_poll_clock_only() {
        let mut server = mockito::Server::new_async().await;
        let good = body(5);
        let etag = etag_for(&good);
        let download = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag)
            .with_body(&good)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        download.assert_async().await;

        let activated_at = h.meta().expect("meta").last_activated_at;
        let before = Arc::as_ptr(&h.handle.load_full());
        h.last_poll_ok_at.store(1, Ordering::Relaxed);

        let revalidate = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(304)
            .with_header("ETag", &etag)
            .create_async()
            .await;
        assert_eq!(h.poller.poll_once().await, PollOutcome::NotModified);
        revalidate.assert_async().await;

        assert!(h.last_poll_ok_at.load(Ordering::Relaxed) > 1);
        assert!(h.last_fetch_ok.load(Ordering::Relaxed));
        assert_eq!(
            before,
            Arc::as_ptr(&h.handle.load_full()),
            "the rule set must not be reloaded on a 304"
        );
        let meta = h.meta().expect("meta");
        assert_eq!(
            meta.last_activated_at, activated_at,
            "the activation tier must not move on a 304"
        );
        assert!(meta.last_poll_ok_at.is_some());
    }

    // -----------------------------------------------------------------------
    // 404 / 5xx / 429
    // -----------------------------------------------------------------------

    /// A transient 404 must never silently disarm a host.
    #[tokio::test]
    async fn not_found_keeps_the_resident_bundle_enforcing() {
        let mut server = mockito::Server::new_async().await;
        let good = body(3);
        let download = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag_for(&good))
            .with_body(&good)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        download.assert_async().await;
        let polled_at = h.last_poll_ok_at.load(Ordering::Relaxed);

        let gone = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(404)
            .with_body(r#"{"error":{"code":"not_found","message":"no bundle"}}"#)
            .create_async()
            .await;
        assert_eq!(h.poller.poll_once().await, PollOutcome::NoBundle);
        gone.assert_async().await;

        assert!(h.still_enforcing(), "a 404 must not disarm the host");
        assert_eq!(h.resident_revision(), Some(3));
        // A permanent 404 must not suppress the OL-1213 staleness warning.
        assert_eq!(
            h.last_poll_ok_at.load(Ordering::Relaxed),
            polled_at,
            "a 404 is neither a 2xx nor a 304"
        );
        assert!(!h.last_fetch_ok.load(Ordering::Relaxed));
    }

    #[tokio::test]
    async fn server_error_keeps_the_resident_bundle_enforcing() {
        let mut server = mockito::Server::new_async().await;
        let good = body(4);
        let download = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag_for(&good))
            .with_body(&good)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        download.assert_async().await;

        let boom = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(503)
            .create_async()
            .await;
        assert_eq!(h.poller.poll_once().await, PollOutcome::Failed);
        boom.assert_async().await;

        assert!(h.still_enforcing());
        assert!(!h.last_fetch_ok.load(Ordering::Relaxed));
        assert_eq!(
            h.meta().expect("meta").etag.as_deref(),
            Some(etag_for(&good).as_str()),
            "a transport failure must not disturb the stored validator"
        );
    }

    #[tokio::test]
    async fn rate_limit_honours_retry_after() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(429)
            .with_header("Retry-After", "45")
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::RateLimited(Some(Duration::from_secs(45)))
        );
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn rate_limit_without_a_usable_header_falls_back_to_the_interval() {
        // An UNPARSABLE header — not a date. The HTTP-date form is legal and
        // is now honoured (see `rate_limit_honours_an_http_date_header`), so it
        // must not be used as the example of an unusable value here.
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(429)
            .with_header("Retry-After", "when we feel like it")
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert_eq!(h.poller.poll_once().await, PollOutcome::RateLimited(None));
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn rate_limit_honours_an_http_date_header() {
        // RFC 9110 §10.2.3 permits an HTTP-date, and rate-limiting origins send
        // it. Treating it as unusable meant retrying at the ordinary interval
        // against an origin that had just said it was overloaded.
        let future = chrono::Utc::now() + chrono::Duration::seconds(900);
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(429)
            .with_header(
                "Retry-After",
                &future.format("%a, %d %b %Y %H:%M:%S GMT").to_string(),
            )
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        match h.poller.poll_once().await {
            PollOutcome::RateLimited(Some(d)) => assert!(
                d >= Duration::from_secs(880) && d <= Duration::from_secs(900),
                "expected ~900s from the http-date, got {d:?}"
            ),
            other => panic!("expected a honoured Retry-After, got {other:?}"),
        }
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn retry_after_is_capped() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(429)
            .with_header("Retry-After", "999999")
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::RateLimited(Some(Duration::from_secs(MAX_RETRY_AFTER_SECS)))
        );
        mock.assert_async().await;
    }

    // -----------------------------------------------------------------------
    // Auth (D46)
    // -----------------------------------------------------------------------

    /// A 401 pauses fetching without terminating the loop, and the pause lifts
    /// as soon as the credential rotates.
    #[tokio::test]
    async fn auth_failure_pauses_until_the_credential_changes() {
        let mut server = mockito::Server::new_async().await;
        let good = body(6);
        let download = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag_for(&good))
            .with_body(&good)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        download.assert_async().await;

        // Exactly ONE 401: the following ticks must not reach the network.
        let revoked = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(401)
            .expect(1)
            .create_async()
            .await;
        assert_eq!(h.poller.poll_once().await, PollOutcome::AuthFailed);
        for _ in 0..3 {
            assert_eq!(
                h.poller.poll_once().await,
                PollOutcome::Skipped("credential_unchanged_after_auth_failure")
            );
        }
        revoked.assert_async().await;

        assert!(
            h.still_enforcing(),
            "a revoked key must not disarm the host"
        );
        assert!(
            !h.cloud_state.is_auth_error(),
            "D46: the poller must never write the cloud auth latch"
        );

        // The cloud worker's credential poll rotates the key — the very next
        // tick fetches again.
        let rotated = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag_for(&good))
            .with_body(&good)
            .expect(1)
            .create_async()
            .await;
        h.credentials.set_key("ol_org_rotated");
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        rotated.assert_async().await;
    }

    /// The cloud worker's latch is honoured — read-only, and never cleared here.
    #[tokio::test]
    async fn latched_cloud_auth_error_skips_the_fetch() {
        let mut server = mockito::Server::new_async().await;
        let never = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .expect(0)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        h.cloud_state.auth_error.store(true, Ordering::Relaxed);
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Skipped("auth_error_latched")
        );
        assert!(
            h.cloud_state.is_auth_error(),
            "the poller must not clear a latch it does not own"
        );
        never.assert_async().await;
    }

    #[tokio::test]
    async fn missing_credential_skips_the_fetch() {
        let mut server = mockito::Server::new_async().await;
        let never = server
            .mock("GET", BUNDLE_ENDPOINT)
            .expect(0)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        *h.credentials.key.lock().expect("lock") = None;
        assert_eq!(
            h.poller.poll_once().await,
            PollOutcome::Skipped("no_credential")
        );
        never.assert_async().await;
    }

    // -----------------------------------------------------------------------
    // Staleness (OL-1213)
    // -----------------------------------------------------------------------

    /// Staleness is a warning, never a disarm — and it is measured from the last
    /// successful **poll**, so a `304` resets it.
    #[tokio::test]
    async fn staleness_warns_and_keeps_enforcing() {
        let mut server = mockito::Server::new_async().await;
        let good = body(8);
        let etag = etag_for(&good);
        let download = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag)
            .with_body(&good)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        download.assert_async().await;
        assert!(!h.poller.check_staleness(), "a fresh poll is not stale");

        // Back-date the poll clock past the threshold.
        h.last_poll_ok_at.store(
            now_unix_secs() - (h.poller.config.stale_warn_after_secs as i64) - 60,
            Ordering::Relaxed,
        );
        assert!(h.poller.check_staleness(), "OL-1213 must fire");
        assert!(
            h.still_enforcing(),
            "staleness must never stop the host enforcing"
        );

        // A 304 is a successful poll and clears the warning — otherwise a
        // healthy org whose rules never change would warn forever.
        let revalidate = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(304)
            .create_async()
            .await;
        assert_eq!(h.poller.poll_once().await, PollOutcome::NotModified);
        revalidate.assert_async().await;
        assert!(!h.poller.check_staleness());
    }

    #[test]
    fn a_host_that_never_polled_successfully_does_not_warn_as_stale() {
        let h = Harness::new("http://127.0.0.1:1".to_string());
        assert_eq!(h.last_poll_ok_at.load(Ordering::Relaxed), 0);
        assert!(!h.poller.check_staleness());
    }

    /// The staleness clock survives a restart: connectivity does not reset
    /// because the daemon did.
    #[tokio::test]
    async fn poll_clock_is_seeded_from_disk() {
        let mut server = mockito::Server::new_async().await;
        let good = body(12);
        let download = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag_for(&good))
            .with_body(&good)
            .expect(1)
            .create_async()
            .await;

        let mut h = Harness::new(server.url());
        assert!(matches!(
            h.poller.poll_once().await,
            PollOutcome::Activated { .. }
        ));
        download.assert_async().await;

        // A fresh poller over the same directory — a restarted daemon.
        let restarted = PolicyPoller::new(
            new_handle(None),
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicI64::new(0)),
            CloudState::new(),
            TestCredentialProvider::with_key("ol_org_test"),
            server.url(),
            PolicyConfig {
                enabled: true,
                poll_interval_secs: 300,
                stale_warn_after_secs: 86_400,
            },
            h.dir.path().to_path_buf(),
            reqwest::Client::new(),
            Some(AGENT_ID.to_string()),
        );
        restarted.seed_poll_clock();
        assert!(restarted.last_poll_ok_at.load(Ordering::Relaxed) > 0);
    }

    // -----------------------------------------------------------------------
    // Boot fetch
    // -----------------------------------------------------------------------

    /// A fresh daemon must be enforcing without waiting a full interval. With a
    /// 3600 s interval, anything that arrives inside a second can only have come
    /// from the boot fetch.
    #[tokio::test]
    async fn boot_fetch_happens_before_the_first_tick() {
        let mut server = mockito::Server::new_async().await;
        let good = body(21);
        let mock = server
            .mock("GET", BUNDLE_ENDPOINT)
            .with_status(200)
            .with_header("ETag", &etag_for(&good))
            .with_body(&good)
            .create_async()
            .await;

        let dir = tempfile::tempdir().expect("tempdir");
        let handle = new_handle(None);
        let task = tokio::spawn(run_policy_poller(
            handle.clone(),
            Arc::new(AtomicBool::new(false)),
            Arc::new(AtomicI64::new(0)),
            CloudState::new(),
            TestCredentialProvider::with_key("ol_org_test"),
            server.url(),
            PolicyConfig {
                enabled: true,
                poll_interval_secs: 3_600,
                stale_warn_after_secs: 86_400,
            },
            dir.path().to_path_buf(),
            reqwest::Client::new(),
            Some(AGENT_ID.to_string()),
        ));

        let mut activated = false;
        for _ in 0..50 {
            if handle.load().is_some() {
                activated = true;
                break;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        task.abort();

        assert!(
            activated,
            "the poller must fetch on boot, not on the first timer tick"
        );
        assert_eq!(
            handle.load().as_ref().as_ref().map(|b| b.revision),
            Some(21)
        );
        mock.assert_async().await;
    }
}