areev 0.2.0

//! HTTP transport layer for the Areev SDK.
//!
//! The public surface here is twofold:
//!
//! 1. **Typed convenience methods** (`add`, `recall`, `get`, …) — these
//!    are the original v0.x surface and remain stable. They are scoped
//!    to the configured `memory_id`.
//! 2. **Generic verb methods** (`_get`, `_post`, `_put`, `_patch`,
//!    `_delete`) — these power the new resource-namespaced API
//!    (`client.tools()`, `client.harness()`, …) and are kept
//!    `pub(crate)` because customers should reach for the resource
//!    layer instead.
//!
//! Both surfaces share the same retry loop ([`HttpClient::send_with_retry`])
//! which retries on transport errors, 5xx, 408, and 429 with jittered
//! backoff: 100 ms, 500 ms, 2 s (± 50 % jitter), up to 3 attempts.

use std::time::{Duration, SystemTime, UNIX_EPOCH};

use crate::error::{AreevError, Result};
use crate::types::*;

const RETRY_DELAYS_MS: [u64; 3] = [100, 500, 2000];

/// Async HTTP transport for Areev.
///
/// Wraps [`reqwest::Client`] with auth, URL building, error dispatch,
/// and a small retry loop. Customers normally reach for
/// [`crate::Areev`] (which embeds an `HttpClient`); construct one
/// directly only when integrating with a custom runtime or test
/// harness.
pub struct HttpClient {
    client: reqwest::Client,
    base_url: String,
    memory_id: String,
    api_key: Option<String>,
    max_retries: usize,
}

/// Production-safe default per-request timeout. Configurable via
/// [`HttpClientBuilder::timeout`].
pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
/// Default TCP connect timeout — a slow DNS / TLS handshake fails fast.
pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);

/// Builder for [`HttpClient`]. Configures TLS, timeouts, retry count, and
/// custom reqwest::Client overrides for enterprise deployments.
///
/// Example:
/// ```no_run
/// use std::time::Duration;
/// use areev::HttpClient;
///
/// let client = HttpClient::builder()
///     .base_url("https://dub.areev.ai")
///     .memory_id("default")
///     .api_key("ar_prod_xyz")
///     .timeout(Duration::from_secs(60))
///     .max_retries(5)
///     .build()
///     .unwrap();
/// ```
pub struct HttpClientBuilder {
    base_url: Option<String>,
    memory_id: Option<String>,
    api_key: Option<String>,
    timeout: Duration,
    connect_timeout: Duration,
    max_retries: usize,
    root_certs: Vec<reqwest::Certificate>,
    danger_invalid_certs: bool,
    pool_max_idle_per_host: Option<usize>,
}

impl Default for HttpClientBuilder {
    fn default() -> Self {
        Self {
            base_url: None,
            memory_id: None,
            api_key: None,
            timeout: DEFAULT_REQUEST_TIMEOUT,
            connect_timeout: DEFAULT_CONNECT_TIMEOUT,
            max_retries: 3,
            root_certs: Vec::new(),
            danger_invalid_certs: false,
            pool_max_idle_per_host: None,
        }
    }
}

impl HttpClientBuilder {
    /// Set the Areev server base URL (no trailing slash required).
    pub fn base_url(mut self, url: impl Into<String>) -> Self {
        self.base_url = Some(url.into());
        self
    }
    /// Set the memory id this client targets.
    pub fn memory_id(mut self, id: impl Into<String>) -> Self {
        self.memory_id = Some(id.into());
        self
    }
    /// Set the API key (`ar_*` or `ap_*`). Sent as `Authorization: Bearer …`.
    pub fn api_key(mut self, key: impl Into<String>) -> Self {
        self.api_key = Some(key.into());
        self
    }
    /// Per-request total timeout. Default 30s. Pass `Duration::MAX` to
    /// disable (NOT recommended — a stuck server will hang the task).
    pub fn timeout(mut self, t: Duration) -> Self {
        self.timeout = t;
        self
    }
    /// TCP-connect timeout. Default 10s.
    pub fn connect_timeout(mut self, t: Duration) -> Self {
        self.connect_timeout = t;
        self
    }
    /// Override max retry attempts on 5xx/408/429/transport. Default 3.
    pub fn max_retries(mut self, n: usize) -> Self {
        self.max_retries = n;
        self
    }
    /// Add a custom trust-root certificate (enterprise CA bundle for
    /// Zscaler/Netskope/PaloAlto TLS-MITM proxies). Multiple calls
    /// accumulate.
    pub fn root_cert(mut self, cert: reqwest::Certificate) -> Self {
        self.root_certs.push(cert);
        self
    }
    /// DANGER: accept self-signed / invalid TLS certs. Dev only — NEVER
    /// in production. FedRAMP / SOC 2 environments will fail audit.
    pub fn danger_accept_invalid_certs(mut self, accept: bool) -> Self {
        self.danger_invalid_certs = accept;
        self
    }
    /// Override the reqwest connection pool size. Default = reqwest's
    /// default (currently unlimited per host).
    pub fn pool_max_idle_per_host(mut self, n: usize) -> Self {
        self.pool_max_idle_per_host = Some(n);
        self
    }

    /// Finalize the builder. Returns an error iff `base_url` or
    /// `memory_id` is unset, or if the underlying reqwest::Client fails
    /// to construct (TLS root-cert chain rejected, etc.).
    pub fn build(self) -> Result<HttpClient> {
        let base_url = self
            .base_url
            .ok_or_else(|| AreevError::Other {
                http_status: 0,
                code: Some("CFG-E001".into()),
                message: "HttpClientBuilder: base_url is required".into(),
                body: serde_json::Value::Null,
                request_id: None,
            })?
            .trim_end_matches('/')
            .to_string();
        let memory_id = self.memory_id.ok_or_else(|| AreevError::Other {
            http_status: 0,
            code: Some("CFG-E002".into()),
            message: "HttpClientBuilder: memory_id is required".into(),
            body: serde_json::Value::Null,
            request_id: None,
        })?;
        let mut cb = reqwest::Client::builder()
            .timeout(self.timeout)
            .connect_timeout(self.connect_timeout);
        for cert in self.root_certs {
            cb = cb.add_root_certificate(cert);
        }
        if self.danger_invalid_certs {
            cb = cb.danger_accept_invalid_certs(true);
        }
        if let Some(n) = self.pool_max_idle_per_host {
            cb = cb.pool_max_idle_per_host(n);
        }
        let client = cb.build().map_err(|e| AreevError::Other {
            http_status: 0,
            code: Some("CFG-E003".into()),
            message: format!("reqwest client build failed: {e}"),
            body: serde_json::Value::Null,
            request_id: None,
        })?;
        Ok(HttpClient {
            client,
            base_url,
            memory_id,
            api_key: self.api_key,
            max_retries: self.max_retries,
        })
    }
}

impl HttpClient {
    /// Construct a builder. Fluent way to set TLS, timeout, retry count,
    /// and custom reqwest options.
    pub fn builder() -> HttpClientBuilder {
        HttpClientBuilder::default()
    }

    /// Construct a client targeting `base_url` and scoped to `memory_id`
    /// with production-safe defaults (30s timeout, 10s connect, 3 retries).
    ///
    /// Use [`HttpClient::builder`] for enterprise customization (custom CA
    /// bundle, longer timeouts, connection-pool tuning).
    pub fn new(base_url: &str, memory_id: &str, api_key: Option<&str>) -> Self {
        Self::builder()
            .base_url(base_url)
            .memory_id(memory_id)
            .api_key(api_key.unwrap_or("").to_string())
            .build()
            .unwrap_or_else(|_| Self {
                // Fallback: blank API key + default-config reqwest::Client.
                // The builder only fails on impossible config; this branch
                // is dead code in practice.
                client: reqwest::Client::new(),
                base_url: base_url.trim_end_matches('/').to_string(),
                memory_id: memory_id.to_string(),
                api_key: api_key.map(|s| s.to_string()),
                max_retries: 3,
            })
    }

    /// Override the maximum number of retry attempts. The default is 3
    /// (i.e. 1 initial attempt plus up to 3 retries).
    pub fn with_max_retries(mut self, max_retries: usize) -> Self {
        self.max_retries = max_retries;
        self
    }

    /// The base URL this client targets (without trailing slash).
    pub fn base_url(&self) -> &str {
        &self.base_url
    }

    /// The memory id this client is scoped to.
    pub fn memory_id(&self) -> &str {
        &self.memory_id
    }

    fn url(&self, path: &str) -> String {
        format!("{}/api/memories/{}/{}", self.base_url, self.memory_id, path)
    }

    /// Build a URL from a leading-slash API path
    /// (e.g. `/memories/{id}/add`). Used by the resource layer.
    fn full_url(&self, path: &str) -> String {
        if let Some(stripped) = path.strip_prefix('/') {
            format!("{}/api/{}", self.base_url, stripped)
        } else {
            format!("{}/api/{}", self.base_url, path)
        }
    }

    fn request(&self, method: reqwest::Method, path: &str) -> reqwest::RequestBuilder {
        let mut req = self.client.request(method, self.url(path));
        if let Some(ref key) = self.api_key {
            req = req.bearer_auth(key);
        }
        req
    }

    fn raw_request(&self, method: reqwest::Method, full_url: &str) -> reqwest::RequestBuilder {
        let mut req = self.client.request(method, full_url);
        if let Some(ref key) = self.api_key {
            req = req.bearer_auth(key);
        }
        req
    }

    /// Emit a DEBUG trace for an outbound resource-layer request.
    ///
    /// `sensitive` requests (DSR exports, audit, provenance, PII
    /// detection, secret rotation) log only the method + path and never
    /// the request/response body, so PII / PHI / secrets stay out of
    /// customer logs (compliance + security guardrail).
    fn log_request(method: &str, path: &str, sensitive: bool) {
        if sensitive {
            tracing::debug!(target: "areev::http", method, path, "areev request (body redacted)");
        } else {
            tracing::debug!(target: "areev::http", method, path, "areev request");
        }
    }

    // ── Generic verb methods (used by the resource layer) ────────────────
    //
    // Underscore-prefixed because the typed convenience methods below
    // shadow plain names. They run every request through
    // `send_with_retry`, which retries on transport errors + 5xx + 408 +
    // 429.

    /// Send a GET request to an absolute API path.
    ///
    /// Low-level escape hatch used by the resource layer. Customers
    /// should prefer [`crate::Areev::system`] / [`crate::Areev::tools`]
    /// / etc. — these verb methods bypass any future typed wrappers.
    pub async fn _get(
        &self,
        path: &str,
        query: Option<&serde_json::Value>,
    ) -> Result<serde_json::Value> {
        Self::log_request("GET", path, false);
        let url = self.full_url(path);
        let url_ref = &url;
        let query_ref = query;
        self.send_with_retry("GET", path, move || {
            let mut req = self.raw_request(reqwest::Method::GET, url_ref);
            if let Some(q) = query_ref {
                req = req.query(q);
            }
            req
        })
        .await
    }

    /// Send a POST request with auto-retry + Idempotency-Key.
    ///
    /// Low-level escape hatch used by the resource layer.
    pub async fn _post(
        &self,
        path: &str,
        body: Option<&serde_json::Value>,
    ) -> Result<serde_json::Value> {
        Self::log_request("POST", path, false);
        let url = self.full_url(path);
        let url_ref = &url;
        let body_ref = body;
        self.send_with_retry("POST", path, move || {
            let req = self.raw_request(reqwest::Method::POST, url_ref);
            match body_ref {
                Some(b) => req.json(b),
                None => req,
            }
        })
        .await
    }

    /// GET that returns the raw response body as bytes (no JSON parse).
    ///
    /// Backs [`crate::resources::Grains::get_raw`], whose endpoint
    /// (`GET /memories/{id}/grains/{hash}/raw`) responds with
    /// `application/octet-stream` — the raw `.mg` blob (9-byte header +
    /// canonical MessagePack), NOT JSON. Retried on transport errors +
    /// 5xx + 408 + 429 like the JSON verbs.
    pub async fn _get_bytes(
        &self,
        path: &str,
        query: Option<&serde_json::Value>,
    ) -> Result<Vec<u8>> {
        Self::log_request("GET", path, false);
        let url = self.full_url(path);
        let idem_key = new_idempotency_key();
        let mut last_err: Option<AreevError> = None;

        for attempt in 0..=self.max_retries {
            let mut req = self.raw_request(reqwest::Method::GET, &url);
            if let Some(q) = query {
                req = req.query(q);
            }
            req = req.header("Idempotency-Key", &idem_key);
            let mut retry_after: Option<Duration> = None;
            match req.send().await {
                Ok(resp) => {
                    if resp.status().as_u16() == 429 {
                        retry_after = parse_retry_after(
                            resp.headers()
                                .get("retry-after")
                                .and_then(|v| v.to_str().ok()),
                        );
                    }
                    match Self::interpret_bytes_response(resp).await {
                        Ok(bytes) => return Ok(bytes),
                        Err(e) => {
                            let retryable = e.is_retryable();
                            last_err = Some(e);
                            if !retryable {
                                return Err(last_err.unwrap());
                            }
                        }
                    }
                }
                Err(e) => {
                    last_err = Some(AreevError::from(e));
                }
            }

            if attempt >= self.max_retries {
                break;
            }
            let sleep_dur = retry_after.unwrap_or_else(|| {
                let base = RETRY_DELAYS_MS[attempt.min(RETRY_DELAYS_MS.len() - 1)];
                Duration::from_millis(jittered_delay_ms(base, attempt))
            });
            tokio::time::sleep(sleep_dur).await;
        }

        Err(last_err.unwrap_or_else(|| AreevError::Other {
            http_status: 0,
            code: None,
            message: "retry loop exited without an error".into(),
            body: serde_json::Value::Null,
            request_id: None,
        }))
    }

    /// POST a raw (non-JSON) string body with an explicit `Content-Type`.
    ///
    /// Backs [`crate::resources::Grains::cal`]: the `/cal` endpoint takes
    /// the CAL query as the raw request body (Content-Type `text/cal`),
    /// the server parses the text into a `CalQuery` AST internally. The
    /// JSON path (`application/json+cal`) instead expects a pre-parsed AST
    /// and rejects a bare string with `CAL-E120`, so this verb is the
    /// ergonomic text path. Retried like the JSON verbs.
    pub async fn _post_raw(
        &self,
        path: &str,
        body: String,
        content_type: &str,
    ) -> Result<serde_json::Value> {
        Self::log_request("POST", path, false);
        let url = self.full_url(path);
        let url_ref = &url;
        let body_ref = &body;
        self.send_with_retry("POST", path, move || {
            self.raw_request(reqwest::Method::POST, url_ref)
                .header("Content-Type", content_type)
                .body(body_ref.clone())
        })
        .await
    }

    /// Send a POST that MUST NOT be retried. Use for single-use payloads
    /// where a retry corrupts server state (OAuth `authorization_code`
    /// per RFC 6749 §4.1.2, financial ops). Does NOT stamp an
    /// `Idempotency-Key` (no retries → no need).
    pub async fn _post_no_retry(
        &self,
        path: &str,
        body: Option<&serde_json::Value>,
    ) -> Result<serde_json::Value> {
        Self::log_request("POST", path, false);
        let url = self.full_url(path);
        let mut req = self.raw_request(reqwest::Method::POST, &url);
        if let Some(b) = body {
            req = req.json(b);
        }
        let resp = req.send().await?;
        Self::interpret_response(resp).await
    }

    /// Send a PUT request to an absolute API path.
    ///
    /// Low-level escape hatch used by the resource layer.
    pub async fn _put(
        &self,
        path: &str,
        body: Option<&serde_json::Value>,
    ) -> Result<serde_json::Value> {
        Self::log_request("PUT", path, false);
        let url = self.full_url(path);
        let url_ref = &url;
        let body_ref = body;
        self.send_with_retry("PUT", path, move || {
            let req = self.raw_request(reqwest::Method::PUT, url_ref);
            match body_ref {
                Some(b) => req.json(b),
                None => req,
            }
        })
        .await
    }

    /// Send a PATCH request to an absolute API path.
    ///
    /// Low-level escape hatch used by the resource layer.
    pub async fn _patch(
        &self,
        path: &str,
        body: Option<&serde_json::Value>,
    ) -> Result<serde_json::Value> {
        Self::log_request("PATCH", path, false);
        let url = self.full_url(path);
        let url_ref = &url;
        let body_ref = body;
        self.send_with_retry("PATCH", path, move || {
            let req = self.raw_request(reqwest::Method::PATCH, url_ref);
            match body_ref {
                Some(b) => req.json(b),
                None => req,
            }
        })
        .await
    }

    /// Send a DELETE request to an absolute API path.
    ///
    /// Low-level escape hatch used by the resource layer.
    pub async fn _delete(&self, path: &str) -> Result<serde_json::Value> {
        Self::log_request("DELETE", path, false);
        let url = self.full_url(path);
        let url_ref = &url;
        self.send_with_retry("DELETE", path, move || {
            self.raw_request(reqwest::Method::DELETE, url_ref)
        })
        .await
    }

    /// DELETE that MUST NOT be retried (irreversible destructive ops:
    /// `namespaces.purge`). A retried purge is a no-op on the second hit
    /// but the caller owns the decision after a transport/5xx failure.
    pub async fn _delete_no_retry(&self, path: &str) -> Result<serde_json::Value> {
        Self::log_request("DELETE", path, false);
        let url = self.full_url(path);
        let req = self.raw_request(reqwest::Method::DELETE, &url);
        let resp = req.send().await?;
        Self::interpret_response(resp).await
    }

    /// GET that suppresses response-body logging — for DSR / provenance /
    /// audit reads whose bodies carry personal data.
    pub async fn _get_sensitive(
        &self,
        path: &str,
        query: Option<&serde_json::Value>,
    ) -> Result<serde_json::Value> {
        Self::log_request("GET", path, true);
        let url = self.full_url(path);
        let url_ref = &url;
        let query_ref = query;
        self.send_with_retry("GET", path, move || {
            let mut req = self.raw_request(reqwest::Method::GET, url_ref);
            if let Some(q) = query_ref {
                req = req.query(q);
            }
            req
        })
        .await
    }

    /// POST that suppresses request/response body logging (DSR verify /
    /// scope-erase / forget-user / PII detection).
    ///
    /// `retry` toggles the bounded auto-retry loop — destructive,
    /// non-idempotent ops (`scope.erase`, `grains.forget_user`) pass
    /// `false`.
    pub async fn _post_sensitive(
        &self,
        path: &str,
        body: Option<&serde_json::Value>,
        retry: bool,
    ) -> Result<serde_json::Value> {
        Self::log_request("POST", path, true);
        if !retry {
            let url = self.full_url(path);
            let mut req = self.raw_request(reqwest::Method::POST, &url);
            if let Some(b) = body {
                req = req.json(b);
            }
            let resp = req.send().await?;
            return Self::interpret_response(resp).await;
        }
        let url = self.full_url(path);
        let url_ref = &url;
        let body_ref = body;
        self.send_with_retry("POST", path, move || {
            let req = self.raw_request(reqwest::Method::POST, url_ref);
            match body_ref {
                Some(b) => req.json(b),
                None => req,
            }
        })
        .await
    }

    /// POST with an optional `If-Match` header and NO auto-retry —
    /// backs `harness.rotate_callback_secret`. The response carries a
    /// one-time secret, so the body is excluded from logging.
    pub async fn _post_if_match_no_retry(
        &self,
        path: &str,
        body: Option<&serde_json::Value>,
        if_match: Option<&str>,
    ) -> Result<serde_json::Value> {
        Self::log_request("POST", path, true);
        let url = self.full_url(path);
        let mut req = self.raw_request(reqwest::Method::POST, &url);
        if let Some(etag) = if_match {
            req = req.header("If-Match", etag);
        }
        if let Some(b) = body {
            req = req.json(b);
        }
        let resp = req.send().await?;
        Self::interpret_response(resp).await
    }

    /// POST a `multipart/form-data` upload — backs `imports.import_file`,
    /// `imports.import_document`, `imports.extract_document`. Not retried
    /// (the file part is consumed on first send).
    pub async fn _post_multipart(
        &self,
        path: &str,
        form: reqwest::multipart::Form,
    ) -> Result<serde_json::Value> {
        Self::log_request("POST", path, false);
        let url = self.full_url(path);
        let req = self
            .raw_request(reqwest::Method::POST, &url)
            .multipart(form);
        let resp = req.send().await?;
        Self::interpret_response(resp).await
    }

    /// Open a Server-Sent-Events stream (POST). Returns an [`SseStream`]
    /// the caller pumps with [`SseStream::next`]. Not auto-retried — a
    /// stream is one live connection; restart on failure.
    pub async fn _stream_sse(
        &self,
        path: &str,
        body: Option<&serde_json::Value>,
    ) -> Result<SseStream> {
        Self::log_request("POST", path, false);
        let url = self.full_url(path);
        let mut req = self
            .raw_request(reqwest::Method::POST, &url)
            .header("Accept", "text/event-stream");
        if let Some(b) = body {
            req = req.json(b);
        }
        let resp = req.send().await?;
        let status = resp.status().as_u16();
        if status >= 400 {
            let request_id = resp
                .headers()
                .get("x-request-id")
                .or_else(|| resp.headers().get("x-areev-request-id"))
                .and_then(|v| v.to_str().ok())
                .map(|s| s.to_string());
            let text = resp.text().await.unwrap_or_default();
            let (message, code, body) = parse_error_body(&text);
            return Err(AreevError::from_response_with_body(
                status,
                code.as_deref(),
                message,
                body,
                request_id,
            ));
        }
        Ok(SseStream::new(resp))
    }

    /// Send with bounded retry on transport errors + 5xx + 408 + 429.
    ///
    /// Stamps an `Idempotency-Key` header (UUIDv4) on every attempt of
    /// the same logical call so the server's idempotency_keys table
    /// dedups duplicate side effects on transient 5xx retries.
    ///
    /// 429 honours the `Retry-After` header (RFC 7231) when present.
    async fn send_with_retry<F>(
        &self,
        _method: &str,
        _path: &str,
        build_request: F,
    ) -> Result<serde_json::Value>
    where
        F: Fn() -> reqwest::RequestBuilder,
    {
        let idem_key = new_idempotency_key();
        let mut last_err: Option<AreevError> = None;

        for attempt in 0..=self.max_retries {
            let req = build_request().header("Idempotency-Key", &idem_key);
            let send_result = req.send().await;
            let mut retry_after: Option<Duration> = None;
            match send_result {
                Ok(resp) => {
                    if resp.status().as_u16() == 429 {
                        retry_after = parse_retry_after(
                            resp.headers()
                                .get("retry-after")
                                .and_then(|v| v.to_str().ok()),
                        );
                    }
                    match Self::interpret_response(resp).await {
                        Ok(value) => return Ok(value),
                        Err(e) => {
                            let retryable = e.is_retryable();
                            last_err = Some(e);
                            if !retryable {
                                return Err(last_err.unwrap());
                            }
                        }
                    }
                }
                Err(e) => {
                    // Transport failure (DNS, refused, TLS, timeout). Retryable.
                    last_err = Some(AreevError::from(e));
                }
            }

            if attempt >= self.max_retries {
                break;
            }
            let sleep_dur = retry_after.unwrap_or_else(|| {
                let base = RETRY_DELAYS_MS[attempt.min(RETRY_DELAYS_MS.len() - 1)];
                Duration::from_millis(jittered_delay_ms(base, attempt))
            });
            tokio::time::sleep(sleep_dur).await;
        }

        Err(last_err.unwrap_or_else(|| AreevError::Other {
            http_status: 0,
            code: None,
            message: "retry loop exited without an error".into(),
            body: serde_json::Value::Null,
            request_id: None,
        }))
    }

    async fn interpret_response(resp: reqwest::Response) -> Result<serde_json::Value> {
        let status = resp.status().as_u16();
        let request_id = resp
            .headers()
            .get("x-request-id")
            .or_else(|| resp.headers().get("x-areev-request-id"))
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());

        if status < 400 {
            if status == 204 {
                return Ok(serde_json::Value::Null);
            }
            let bytes = resp.bytes().await?;
            if bytes.is_empty() {
                return Ok(serde_json::Value::Null);
            }
            let v: serde_json::Value = serde_json::from_slice(&bytes)?;
            return Ok(v);
        }

        // Error path — parse body for {error|message, code}.
        let text = resp.text().await.unwrap_or_default();
        let (message, code, body) = parse_error_body(&text);
        Err(AreevError::from_response_with_body(
            status,
            code.as_deref(),
            message,
            body,
            request_id,
        ))
    }

    /// Like [`Self::interpret_response`] but returns the raw body bytes on
    /// success instead of parsing JSON. Error bodies are still parsed for
    /// `{error|message, code}` (servers emit JSON errors regardless of the
    /// success-path content type).
    async fn interpret_bytes_response(resp: reqwest::Response) -> Result<Vec<u8>> {
        let status = resp.status().as_u16();
        let request_id = resp
            .headers()
            .get("x-request-id")
            .or_else(|| resp.headers().get("x-areev-request-id"))
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());

        if status < 400 {
            if status == 204 {
                return Ok(Vec::new());
            }
            return Ok(resp.bytes().await?.to_vec());
        }

        let text = resp.text().await.unwrap_or_default();
        let (message, code, body) = parse_error_body(&text);
        Err(AreevError::from_response_with_body(
            status,
            code.as_deref(),
            message,
            body,
            request_id,
        ))
    }

    // ── Typed convenience methods (memory-scoped) ───────────────────────

    /// Query (recall) grains from the configured memory.
    pub async fn recall(&self, req: &RecallRequest) -> Result<RecallResponse> {
        let resp = self
            .request(reqwest::Method::POST, "recall")
            .json(req)
            .send()
            .await?;
        self.handle_typed(resp).await
    }

    /// Ingest natural language text — creates an Observation grain and
    /// (sync or async) extracts Fact grains.
    pub async fn remember(&self, req: &RememberRequest) -> Result<RememberResponse> {
        let resp = self
            .request(reqwest::Method::POST, "remember")
            .json(req)
            .send()
            .await?;
        self.handle_typed(resp).await
    }

    /// Run one harness turn (Flow-A). Returns a [`HarnessChatResponse`]
    /// whose `status` is either `Completed` (terminal) or
    /// `RequiresAction` (paused on `client://` tools — caller must
    /// execute them and call [`HttpClient::harness_chat_resume`]).
    pub async fn harness_chat(
        &self,
        slug: &str,
        req: &HarnessChatRequest,
    ) -> Result<HarnessChatResponse> {
        let resp = self
            .request(reqwest::Method::POST, &format!("harnesses/{slug}/chat"))
            .json(req)
            .send()
            .await?;
        self.handle_typed(resp).await
    }

    /// Resume a paused harness turn with tool outputs. All
    /// `pending_tool_calls[*].tool_call_id` values from the prior
    /// `requires_action` response must appear exactly once in
    /// `req.tool_outputs`.
    pub async fn harness_chat_resume(
        &self,
        slug: &str,
        req: &ChatResumeRequest,
    ) -> Result<HarnessChatResponse> {
        let resp = self
            .request(
                reqwest::Method::POST,
                &format!("harnesses/{slug}/chat/resume"),
            )
            .json(req)
            .send()
            .await?;
        self.handle_typed(resp).await
    }

    async fn handle_typed<T: serde::de::DeserializeOwned>(
        &self,
        resp: reqwest::Response,
    ) -> Result<T> {
        let status = resp.status().as_u16();
        let request_id = resp
            .headers()
            .get("x-request-id")
            .or_else(|| resp.headers().get("x-areev-request-id"))
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());

        if resp.status().is_success() {
            if status == 204 {
                // Best-effort empty deserialise — most T won't match, callers
                // that hit 204 should use the verb methods above instead.
                return serde_json::from_value(serde_json::Value::Null).map_err(AreevError::from);
            }
            let bytes = resp.bytes().await?;
            if bytes.is_empty() {
                return serde_json::from_value(serde_json::Value::Null).map_err(AreevError::from);
            }
            return serde_json::from_slice(&bytes).map_err(AreevError::from);
        }

        let text = resp.text().await.unwrap_or_default();
        let (message, code, body) = parse_error_body(&text);
        Err(AreevError::from_response_with_body(
            status,
            code.as_deref(),
            message,
            body,
            request_id,
        ))
    }
}

/// Extract `(message, code, body)` from an Areev error response, falling
/// back to the raw text when the body isn't JSON. `body` is the full
/// parsed JSON object (or [`serde_json::Value::Null`] when the body
/// wasn't a JSON object) so the typed error can expose structured detail
/// (`retry_after_ms`, per-field validation errors). Redaction happens
/// downstream in [`AreevError::from_response_with_body`].
fn parse_error_body(text: &str) -> (String, Option<String>, serde_json::Value) {
    if text.is_empty() {
        return ("(empty body)".to_string(), None, serde_json::Value::Null);
    }
    match serde_json::from_str::<serde_json::Value>(text) {
        Ok(value @ serde_json::Value::Object(_)) => {
            let map = value.as_object().expect("matched Object");
            let code = map
                .get("code")
                .and_then(|v| v.as_str())
                .map(|s| s.to_string());
            let message = map
                .get("error")
                .and_then(|v| v.as_str())
                .or_else(|| map.get("message").and_then(|v| v.as_str()))
                .map(|s| s.to_string())
                .unwrap_or_else(|| text.to_string());
            (message, code, value)
        }
        _ => (text.to_string(), None, serde_json::Value::Null),
    }
}

/// Deterministic ±50 % jitter on top of a base delay.
///
/// We derive the perturbation from the system nanosecond clock + the
/// attempt index + the task-local thread id, so the SDK stays
/// dependency-free (no `rand` crate) while spreading retries across
/// concurrent callers (the audit flagged a thundering-herd when only
/// subsec_nanos was mixed in).
fn jittered_delay_ms(base_ms: u64, attempt: usize) -> u64 {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.subsec_nanos() as u64)
        .unwrap_or(0);
    // Mix thread id so two tokio tasks waking at the same instant get
    // different jitter even with the same attempt index.
    let tid = thread_id_u64();
    let mixed = nanos
        .wrapping_add((attempt as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15))
        .wrapping_add(tid.wrapping_mul(0xBF58_476D_1CE4_E5B9));
    let half = base_ms.max(1);
    let perturb = mixed % (half.saturating_mul(2).max(1));
    let centered = perturb as i64 - half as i64;
    let result = base_ms as i64 + centered;
    result.max(0) as u64
}

fn thread_id_u64() -> u64 {
    use std::hash::{Hash, Hasher};
    let mut h = std::collections::hash_map::DefaultHasher::new();
    std::thread::current().id().hash(&mut h);
    h.finish()
}

/// Mint a fresh `Idempotency-Key` for a retryable mutating request.
///
/// A CSPRNG-backed UUIDv4 (128 bits, `getrandom`) rendered as 32 hex
/// chars — matching the Python SDK (`uuid4().hex`) and TS
/// (`crypto.randomUUID`). The key MUST be cryptographically random:
/// the server dedups by this value, so a predictable / collision-prone
/// key across distinct clients would silently coalesce two different
/// writes into one (data loss). The earlier `DefaultHasher(nanos, tid)`
/// scheme is not collision-safe and was replaced.
fn new_idempotency_key() -> String {
    uuid::Uuid::new_v4().simple().to_string()
}

/// Max `Retry-After` we honour (seconds). A malicious or buggy server
/// can't pin a retry arbitrarily far into the future. Mirrors the
/// Python/TS SDK clamp.
const RETRY_AFTER_MAX_SECS: u64 = 300;

/// Parse an RFC 7231 §7.1.3 `Retry-After` header. Accepts either the
/// delta-seconds form (`120`) or the HTTP-date form
/// (`Wed, 21 Oct 2015 07:28:00 GMT`); for a date we honour the delay
/// until that instant. Clamped to `[0, 300]` s so a server can't pin us.
/// Returns `None` on malformed input. Matches Python/TS, which both
/// parse the date form.
fn parse_retry_after(header: Option<&str>) -> Option<Duration> {
    let raw = header?.trim();
    if let Ok(secs) = raw.parse::<u64>() {
        return Some(Duration::from_secs(secs.min(RETRY_AFTER_MAX_SECS)));
    }
    let target = httpdate::parse_http_date(raw).ok()?;
    let delay = target
        .duration_since(SystemTime::now())
        .unwrap_or(Duration::ZERO);
    Some(Duration::from_secs(
        delay.as_secs().min(RETRY_AFTER_MAX_SECS),
    ))
}

/// A live Server-Sent-Events stream returned by [`HttpClient::_stream_sse`].
///
/// Pump it with [`SseStream::next`], which resolves to the next `data:`
/// payload (the bare value, `data: ` prefix stripped) until the stream
/// ends (`Ok(None)`). The terminal `data: [DONE]` sentinel is consumed
/// internally and surfaces as end-of-stream.
///
/// The stream is a single live connection and is **not** auto-retried —
/// restart it on error. Used by [`crate::resources::Chat::stream`].
pub struct SseStream {
    resp: reqwest::Response,
    buf: String,
    pending: std::collections::VecDeque<String>,
    done: bool,
}

impl SseStream {
    fn new(resp: reqwest::Response) -> Self {
        Self {
            resp,
            buf: String::new(),
            pending: std::collections::VecDeque::new(),
            done: false,
        }
    }

    /// Pull the next SSE `data:` payload, or `Ok(None)` at end of stream.
    pub async fn next(&mut self) -> Result<Option<String>> {
        loop {
            if let Some(item) = self.pending.pop_front() {
                return Ok(Some(item));
            }
            if self.done {
                return Ok(None);
            }
            match self.resp.chunk().await? {
                Some(bytes) => {
                    self.buf.push_str(&String::from_utf8_lossy(&bytes));
                    self.drain_frames();
                }
                None => {
                    // Flush any trailing frame without a terminating blank line.
                    self.done = true;
                    if !self.buf.trim().is_empty() {
                        let leftover = std::mem::take(&mut self.buf);
                        self.parse_frame(&leftover);
                    }
                }
            }
        }
    }

    /// Split the buffer on SSE event boundaries (blank line) and parse
    /// each complete frame, leaving any partial frame in the buffer.
    fn drain_frames(&mut self) {
        while let Some(idx) = self.buf.find("\n\n").or_else(|| self.buf.find("\r\n\r\n")) {
            let sep_len = if self.buf[idx..].starts_with("\r\n\r\n") {
                4
            } else {
                2
            };
            let frame = self.buf[..idx].to_string();
            self.buf.drain(..idx + sep_len);
            self.parse_frame(&frame);
        }
    }

    /// Extract `data:` lines from one SSE frame. A `[DONE]` sentinel
    /// closes the stream.
    fn parse_frame(&mut self, frame: &str) {
        for line in frame.lines() {
            let line = line.strip_prefix("data:").map(|s| s.trim_start());
            if let Some(payload) = line {
                if payload == "[DONE]" {
                    self.done = true;
                    return;
                }
                self.pending.push_back(payload.to_string());
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{new_idempotency_key, parse_retry_after, RETRY_AFTER_MAX_SECS};
    use std::collections::HashSet;
    use std::time::Duration;

    #[test]
    fn idempotency_keys_are_unique_across_calls() {
        // CSPRNG UUIDv4 → no collisions across many mints (the old
        // DefaultHasher(nanos, tid) scheme could collide within a tight
        // loop and silently dedup two distinct server writes).
        let n = 10_000;
        let mut seen = HashSet::with_capacity(n);
        for _ in 0..n {
            let k = new_idempotency_key();
            assert_eq!(k.len(), 32, "expected 32 hex chars, got {k:?}");
            assert!(k.chars().all(|c| c.is_ascii_hexdigit()));
            assert!(seen.insert(k), "duplicate idempotency key minted");
        }
    }

    #[test]
    fn parse_retry_after_integer_seconds() {
        assert_eq!(parse_retry_after(Some("3")), Some(Duration::from_secs(3)));
        assert_eq!(
            parse_retry_after(Some("  10 ")),
            Some(Duration::from_secs(10))
        );
        // Clamp to the max.
        assert_eq!(
            parse_retry_after(Some("100000")),
            Some(Duration::from_secs(RETRY_AFTER_MAX_SECS))
        );
    }

    #[test]
    fn parse_retry_after_http_date_form() {
        // A date far in the past → zero delay (not None).
        assert_eq!(
            parse_retry_after(Some("Wed, 21 Oct 2015 07:28:00 GMT")),
            Some(Duration::ZERO)
        );
        // A date far in the future → clamped to the max.
        assert_eq!(
            parse_retry_after(Some("Fri, 31 Dec 9999 23:59:59 GMT")),
            Some(Duration::from_secs(RETRY_AFTER_MAX_SECS))
        );
    }

    #[test]
    fn parse_retry_after_rejects_garbage_and_empty() {
        assert_eq!(parse_retry_after(None), None);
        assert_eq!(parse_retry_after(Some("not-a-date")), None);
        assert_eq!(parse_retry_after(Some("")), None);
    }
}