faucet_source_rest/

stream.rs

//! The main REST stream executor.

use crate::auth::Auth;
use crate::auth::oauth2::TokenCache;
use crate::auth::token_endpoint::TokenEndpointCache;
use crate::config::{RestStreamConfig, TlsClientConfig};
use crate::extract;
use crate::pagination::{PaginationState, PaginationStyle};
use crate::retry;
use async_trait::async_trait;
use faucet_core::replication::{
    BindTarget, ReplicationMethod, filter_incremental, max_replication_value, max_value,
};
use faucet_core::schema;
use faucet_core::{AuthSpec, Credential, CredentialPlacement, FaucetError, SharedAuthProvider};
use futures_core::Stream;
use reqwest::Client;
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
use serde::Deserialize;
use serde_json::Value;
use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex as AsyncMutex;

/// A configured REST API stream that handles pagination, auth, and extraction.
pub struct RestStream {
    config: RestStreamConfig,
    client: Client,
    /// Shared OAuth2 token cache (only used when `config.auth` is `Auth::OAuth2`).
    token_cache: TokenCache,
    /// Shared token endpoint cache (only used when `config.auth` is `Auth::TokenEndpoint`).
    token_endpoint_cache: TokenEndpointCache,
    /// Optional shared auth provider. Set when `config.auth` is an
    /// `AuthSpec::Reference` resolved by the caller (e.g. the CLI `auth:`
    /// catalog), or injected directly by a library caller to share one token
    /// across multiple sources. When present it takes precedence over inline
    /// auth.
    auth_provider: Option<SharedAuthProvider>,
    /// Bookmark applied at runtime via
    /// [`Source::apply_start_bookmark`](faucet_core::Source::apply_start_bookmark).
    /// Takes precedence over `config.start_replication_value` when set.
    runtime_start: Arc<AsyncMutex<Option<Value>>>,
    /// Rendered lower/upper bounds for the current datetime window (#527),
    /// applied to each request by [`execute_request_once`](Self::execute_request_once)
    /// alongside any [`replication_bind`](RestStreamConfig::replication_bind). Set
    /// by the window loop in `stream_pages_inner` before each window's pages;
    /// empty when no `window:` block is configured. Each entry is
    /// `(target, name, rendered-value)`.
    window_binds: Arc<AsyncMutex<Vec<(BindTarget, String, String)>>>,
    /// Test-only override for the "now" upper bound of datetime window slicing
    /// (#527). `None` in production (uses `Utc::now()`); set by unit tests so the
    /// window enumeration is deterministic.
    now_override: Option<chrono::DateTime<chrono::Utc>>,
    /// Retry policy for transient request failures. Built in `new()` from the
    /// REST source's own `config.max_retries` / `config.retry_backoff`. Fed into
    /// the REST `retry::execute_with_retry` runner (which keeps its 429 /
    /// `Retry-After` handling). Overridable via
    /// [`with_retry_policy`](Self::with_retry_policy) — but the REST connector's
    /// own legacy `max_retries` / `retry_backoff` fields take precedence when the
    /// user has set them away from their defaults.
    retry_policy: faucet_core::RetryPolicy,
    /// Static request headers (from `config.headers`, #539), validated once in
    /// [`new`](Self::new) into a [`HeaderMap`] and merged into **every** request
    /// (data pages, async-job requests, `$metadata` probes) *before* the auth
    /// provider's placements — so an auth header of the same name wins.
    static_headers: HeaderMap,
}

/// Default value of [`RestStreamConfig::max_retries`]. When the user leaves this
/// untouched, an injected [`RetryPolicy`](faucet_core::RetryPolicy) is allowed to
/// override it (see [`RestStream::with_retry_policy`]).
const DEFAULT_MAX_RETRIES: u32 = 3;
/// Default value of [`RestStreamConfig::retry_backoff`]. Same precedence rule as
/// [`DEFAULT_MAX_RETRIES`].
const DEFAULT_RETRY_BACKOFF: Duration = Duration::from_secs(1);

/// Attach a mutual-TLS client identity (from [`TlsClientConfig`]) to the HTTP
/// client builder. Only compiled with the `mtls` feature; the non-`mtls` stub
/// errors so a `tls:` block on a build without the feature fails loudly rather
/// than silently sending no client certificate.
#[cfg(feature = "mtls")]
fn apply_client_tls(
    builder: reqwest::ClientBuilder,
    tls: &TlsClientConfig,
) -> Result<reqwest::ClientBuilder, FaucetError> {
    let identity = build_identity(tls)?;
    // Use the native-tls backend explicitly: the identity is built with
    // native-tls constructors, and the workspace may also have rustls compiled
    // in (feature unification) which would otherwise be selected.
    let mut builder = builder.identity(identity).use_native_tls();
    if let Some(v) = &tls.min_version {
        // `TlsClientConfig::validate` guarantees `v` is "1.2" or "1.3".
        let version = if v == "1.3" {
            reqwest::tls::Version::TLS_1_3
        } else {
            reqwest::tls::Version::TLS_1_2
        };
        builder = builder.min_tls_version(version);
    }
    Ok(builder)
}

#[cfg(not(feature = "mtls"))]
fn apply_client_tls(
    _builder: reqwest::ClientBuilder,
    _tls: &TlsClientConfig,
) -> Result<reqwest::ClientBuilder, FaucetError> {
    Err(FaucetError::Config(
        "a `tls:` (mutual-TLS) block is configured, but this build of \
         faucet-source-rest lacks the `mtls` feature; rebuild with \
         `--features mtls`"
            .into(),
    ))
}

/// Build a [`reqwest::Identity`] from the PEM pair or the PKCS#12 file. Errors
/// never echo key material — only the backend's opaque parse message.
#[cfg(feature = "mtls")]
fn build_identity(tls: &TlsClientConfig) -> Result<reqwest::Identity, FaucetError> {
    if let Some(p12_path) = &tls.client_identity_pkcs12 {
        let der = std::fs::read(p12_path).map_err(|e| {
            FaucetError::Config(format!(
                "tls: could not read PKCS#12 file {p12_path:?}: {e}"
            ))
        })?;
        let password = tls.pkcs12_password.as_deref().unwrap_or("");
        reqwest::Identity::from_pkcs12_der(&der, password)
            .map_err(|e| FaucetError::Config(format!("tls: invalid PKCS#12 identity: {e}")))
    } else {
        // `validate()` guarantees both are present on the PEM path.
        let cert = tls.client_cert.as_deref().unwrap_or_default();
        let key = tls.client_key.as_deref().unwrap_or_default();
        reqwest::Identity::from_pkcs8_pem(cert.as_bytes(), key.as_bytes())
            .map_err(|e| FaucetError::Config(format!("tls: invalid PEM client identity: {e}")))
    }
}

/// Map a [`Credential`] from a shared provider onto the REST [`Auth`]
/// representation so the existing header-application path can be reused.
/// Substitute `${name}` tokens with flow-captured login values (#567). Only
/// exact `${name}` occurrences for a captured `name` are replaced; any other
/// `${...}` is left untouched. Applied to the URL and config header values so a
/// captured session value can travel there per request.
fn substitute_captured(s: &str, captured: &std::collections::BTreeMap<String, String>) -> String {
    if captured.is_empty() || !s.contains("${") {
        return s.to_string();
    }
    let mut out = s.to_string();
    for (k, v) in captured {
        out = out.replace(&format!("${{{k}}}"), v);
    }
    out
}

fn credential_to_auth(cred: Credential) -> Auth {
    match cred {
        Credential::Bearer(token) => Auth::Bearer { token },
        Credential::Token(token) => Auth::Custom {
            headers: std::iter::once(("Authorization".to_string(), token)).collect(),
        },
        Credential::Basic { username, password } => Auth::Basic { username, password },
        Credential::Header { name, value } => Auth::Custom {
            headers: std::iter::once((name, value)).collect(),
        },
    }
}

/// First JSONPath match rendered as a string (string verbatim, number as text).
/// Used by the async-job runner to read the job id / status from responses.
fn jsonpath_first_string(v: &Value, path: &str) -> Option<String> {
    use jsonpath_rust::JsonPath;
    let results = v.query(path).ok()?;
    match results.first()? {
        Value::String(s) => Some(s.clone()),
        Value::Number(n) => Some(n.to_string()),
        Value::Bool(b) => Some(b.to_string()),
        _ => None,
    }
}

/// First JSONPath match as an owned [`Value`] (type-preserving). Used by the
/// resumable-cursor bookmark (#547) so a numeric cursor stays a number.
fn jsonpath_first_value(v: &Value, path: &str) -> Option<Value> {
    use jsonpath_rust::JsonPath;
    v.query(path).ok()?.first().map(|x| (*x).clone())
}

/// A locator value counts as "no more pages" when it is empty or the literal
/// string `null` (Salesforce Bulk sends `Sforce-Locator: null` when done).
fn is_terminal_locator(value: &str) -> bool {
    let v = value.trim();
    v.is_empty() || v.eq_ignore_ascii_case("null")
}

/// Read the next result-set locator (#557) from the fetch response header or
/// body, per the `fetch` config. Returns `None` when no locator source is
/// configured or the locator signals completion.
fn next_locator(
    headers: &HeaderMap,
    body: Option<&Value>,
    job: &crate::async_job::AsyncJobConfig,
) -> Option<String> {
    if let Some(name) = &job.fetch.locator_header
        && let Some(raw) = headers.get(name).and_then(|v| v.to_str().ok())
        && !is_terminal_locator(raw)
    {
        return Some(raw.trim().to_string());
    }
    if let Some(path) = &job.fetch.locator_body
        && let Some(body) = body
        && let Some(raw) = jsonpath_first_string(body, path)
        && !is_terminal_locator(&raw)
    {
        return Some(raw.trim().to_string());
    }
    None
}

/// Insert a header from string parts, mapping invalid names/values to a typed
/// config error rather than panicking.
fn insert_header(headers: &mut HeaderMap, name: &str, value: &str) -> Result<(), FaucetError> {
    let hn = HeaderName::from_bytes(name.as_bytes())
        .map_err(|e| FaucetError::Config(format!("rest: invalid header name '{name}': {e}")))?;
    let hv = HeaderValue::from_str(value).map_err(|e| {
        FaucetError::Config(format!("rest: invalid value for header '{name}': {e}"))
    })?;
    headers.insert(hn, hv);
    Ok(())
}

impl RestStream {
    /// Create a new stream from the given configuration.
    pub fn new(mut config: RestStreamConfig) -> Result<Self, FaucetError> {
        // Derive OData request defaults (paging, `$.value`, query sugar, Prefer)
        // before validation so the checks see the effective request shape.
        config.apply_odata_defaults();
        // Cross-field config invariants (e.g. file response formats can't paginate).
        config.validate()?;
        // Validate expiry_ratio at construction time.
        let expiry_ratio_to_validate = match &config.auth {
            AuthSpec::Inline(Auth::OAuth2 { expiry_ratio, .. })
            | AuthSpec::Inline(Auth::TokenEndpoint { expiry_ratio, .. }) => Some(*expiry_ratio),
            _ => None,
        };
        if let Some(ratio) = expiry_ratio_to_validate
            && (ratio <= 0.0 || ratio > 1.0)
        {
            return Err(FaucetError::Auth(format!(
                "expiry_ratio must be in (0.0, 1.0], got {ratio}"
            )));
        }

        let mut builder = Client::builder();
        if let Some(t) = config.timeout {
            builder = builder.timeout(t);
        }
        // Mutual TLS: attach a client certificate/identity to the shared client
        // so it is presented on every request — data pages AND any inline auth
        // token request (both use `self.client`).
        if let Some(tls) = &config.tls {
            tls.validate()?;
            builder = apply_client_tls(builder, tls)?;
        }
        // Build the default retry policy from REST's own legacy reliability
        // fields so behavior is unchanged when no policy is injected. The REST
        // `retry::execute_with_retry` runner is driven by `max_retries`
        // (retries-after-first) + `base`, so `max_attempts = max_retries + 1`.
        let retry_policy = faucet_core::RetryPolicy {
            max_attempts: config.max_retries.saturating_add(1),
            backoff: faucet_core::BackoffKind::Exponential,
            base: config.retry_backoff,
            ..faucet_core::RetryPolicy::default()
        };
        // Static custom headers (#539): validated once here (also validated in
        // `config.validate()` above, so this cannot fail) and reused per request.
        let static_headers = crate::config::build_header_map(&config.headers)?;
        Ok(Self {
            config,
            client: builder.build()?,
            token_cache: TokenCache::new(),
            token_endpoint_cache: TokenEndpointCache::new(),
            auth_provider: None,
            runtime_start: Arc::new(AsyncMutex::new(None)),
            window_binds: Arc::new(AsyncMutex::new(Vec::new())),
            now_override: None,
            retry_policy,
            static_headers,
        })
    }

    /// Attach a shared [`AuthProvider`](faucet_core::AuthProvider). When set, the
    /// provider supplies the credential for every request (taking precedence
    /// over inline auth), so several sources can share one token with
    /// single-flight refresh. Used by the CLI to resolve `auth: { ref }`, and by
    /// library callers who construct one provider and inject it into many
    /// sources.
    pub fn with_auth_provider(mut self, provider: SharedAuthProvider) -> Self {
        self.auth_provider = Some(provider);
        self
    }

    /// Test-only: pin the "now" upper bound used by datetime window slicing (#527)
    /// to a fixed RFC 3339 instant, so the window enumeration is deterministic in
    /// tests. No effect in production (which uses `Utc::now()`). Hidden from docs;
    /// takes a string so callers need not depend on `chrono`.
    #[doc(hidden)]
    pub fn with_now_override_rfc3339(mut self, rfc3339: &str) -> Self {
        self.now_override = chrono::DateTime::parse_from_rfc3339(rfc3339)
            .ok()
            .map(|d| d.with_timezone(&chrono::Utc));
        self
    }

    /// Attach a custom [`RetryPolicy`](faucet_core::RetryPolicy) for transient
    /// request failures, used by the CLI to inject a pipeline-level
    /// `resilience:` policy.
    ///
    /// **Legacy-field precedence:** the REST connector predates the unified
    /// resilience policy and exposes its own `max_retries` / `retry_backoff`
    /// config fields. If the user has set either of those away from its default
    /// (`max_retries: 3`, `retry_backoff: 1s`), those explicit values win and the
    /// injected `policy` is ignored — an explicit per-connector setting is never
    /// silently overridden by a pipeline-wide default. When both fields are at
    /// their defaults, the injected policy takes effect.
    ///
    /// **Inert fields on REST:** because the REST source keeps its own
    /// `429`/`Retry-After`-aware retry runner, it honors only the injected
    /// policy's `max_attempts` (→ `max_retries`) and `base` (→ `retry_backoff`).
    /// The policy's `max` (per-sleep cap), `jitter`, and `retry_on` fields are
    /// **not** honored here — they apply on the `xml`/`graphql` sources and on
    /// every sink-side write.
    pub fn with_retry_policy(mut self, policy: faucet_core::RetryPolicy) -> Self {
        let user_changed_legacy_fields = self.config.max_retries != DEFAULT_MAX_RETRIES
            || self.config.retry_backoff != DEFAULT_RETRY_BACKOFF;
        if !user_changed_legacy_fields {
            self.retry_policy = policy;
        }
        self
    }

    /// Fetch all records across all pages as raw JSON values.
    ///
    /// When `partitions` are configured, the stream is executed once per
    /// partition and all results are concatenated.
    ///
    /// When `replication_method` is `Incremental` and `replication_key` +
    /// `start_replication_value` are both set, records at or before the
    /// bookmark are filtered out.
    pub async fn fetch_all(&self) -> Result<Vec<Value>, FaucetError> {
        if self.config.partitions.is_empty() {
            self.fetch_partition(None, None).await
        } else if let Some(concurrency) = self.config.partition_concurrency {
            // Process partitions concurrently using a semaphore to limit parallelism.
            let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(concurrency.max(1)));
            let mut handles = Vec::with_capacity(self.config.partitions.len());

            for ctx in &self.config.partitions {
                let permit =
                    semaphore.clone().acquire_owned().await.map_err(|e| {
                        FaucetError::Config(format!("semaphore acquire failed: {e}"))
                    })?;
                let fut = self.fetch_partition(Some(ctx), None);
                handles.push(async move {
                    let result = fut.await;
                    drop(permit);
                    result
                });
            }

            let results = futures::future::try_join_all(handles).await?;
            Ok(results.into_iter().flatten().collect())
        } else {
            let mut all_records = Vec::new();
            for ctx in &self.config.partitions {
                let records = self.fetch_partition(Some(ctx), None).await?;
                all_records.extend(records);
            }
            Ok(all_records)
        }
    }

    /// Fetch all records and deserialize into typed structs.
    pub async fn fetch_all_as<T: for<'de> Deserialize<'de>>(&self) -> Result<Vec<T>, FaucetError> {
        let values = self.fetch_all().await?;
        values
            .into_iter()
            .map(|v| serde_json::from_value(v).map_err(FaucetError::Json))
            .collect()
    }

    /// Infer a JSON Schema for this stream's records.
    ///
    /// If a `schema` is already set on the config, it is returned immediately
    /// without making any HTTP requests.
    ///
    /// Otherwise the stream fetches up to `schema_sample_size` records
    /// (respecting `max_pages`) and derives a JSON Schema from them.  Fields
    /// that are absent in some records, or that carry a `null` value, are
    /// marked as nullable (`["<type>", "null"]`).
    ///
    /// Set `schema_sample_size` to `0` to sample all available records.
    pub async fn infer_schema(&self) -> Result<Value, FaucetError> {
        if let Some(ref s) = self.config.schema {
            return Ok(s.clone());
        }
        let limit = match self.config.schema_sample_size {
            0 => None,
            n => Some(n),
        };
        let records = self.fetch_partition(None, limit).await?;
        Ok(schema::infer_schema(&records))
    }

    /// Fetch all records in incremental mode, returning the records along with
    /// the maximum value of `replication_key` observed across those records.
    ///
    /// The returned bookmark should be persisted by the caller and passed back
    /// as `start_replication_value` on the next run.
    ///
    /// If no `replication_key` is configured, this behaves identically to
    /// [`fetch_all`](Self::fetch_all) and the bookmark is `None`.
    pub async fn fetch_all_incremental(&self) -> Result<(Vec<Value>, Option<Value>), FaucetError> {
        let records = self.fetch_all().await?;
        let bookmark = self
            .config
            .replication_key
            .as_deref()
            .and_then(|key| max_replication_value(&records, key))
            .cloned();
        Ok((records, bookmark))
    }

    /// Stream API pages without buffering the full result set.
    ///
    /// This is a thin convenience wrapper around the
    /// [`Source::stream_pages`](faucet_core::Source::stream_pages) trait
    /// method — it discards bookmarks and yields one `Vec<Value>` per
    /// upstream API page. Use the trait method directly if you need
    /// per-page bookmarks for incremental replication.
    ///
    /// Note: this inherent convenience method does not fan out over
    /// `partitions`. The `Source::stream_pages` trait impl (what the pipeline
    /// drives) and [`fetch_all`](Self::fetch_all) do handle multi-partition
    /// streams (#535).
    ///
    /// ```rust,no_run
    /// use faucet_source_rest::{RestStream, RestStreamConfig};
    /// use futures::StreamExt;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let stream = RestStream::new(RestStreamConfig::new("https://api.example.com", "/items"))?;
    /// let mut pages = stream.stream_pages();
    /// while let Some(page) = pages.next().await {
    ///     let records = page?;
    ///     println!("got {} records", records.len());
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn stream_pages(
        &self,
    ) -> Pin<Box<dyn Stream<Item = Result<Vec<Value>, FaucetError>> + Send + '_>> {
        let mut inner = self.stream_pages_inner(None);
        Box::pin(async_stream::try_stream! {
            loop {
                let page = std::future::poll_fn(|cx| inner.as_mut().poll_next(cx)).await;
                match page {
                    Some(Ok(p)) => yield p.records,
                    Some(Err(e)) => Err(e)?,
                    None => break,
                }
            }
        })
    }

    // ── Private helpers ───────────────────────────────────────────────────────

    /// Extract a page's records from a parsed response body, honouring the
    /// configured extraction mode: `records_multi` (#548, op-stamped multi-array
    /// fan-out), `record_ancestors` (#549, nested path with lifted ancestor
    /// fields), or the classic single `records_path`.
    fn extract_page(&self, body: &Value) -> Result<Vec<Value>, FaucetError> {
        extract::extract_configured(
            body,
            self.config.records_path.as_deref(),
            self.config.record_ancestors.as_ref(),
            &self.config.records_multi,
            self.config.op_field.as_deref().unwrap_or("_op"),
        )
    }

    /// Core pagination loop shared by [`Source::stream_pages`] and
    /// [`fetch_partition`](Self::fetch_partition).
    ///
    /// Yields one [`faucet_core::StreamPage`] per page. The final page carries
    /// the consolidated replication bookmark (`Some(value)`); all intermediate
    /// pages carry `None`. When `context` is `Some`, path placeholders are
    /// substituted for partition support.
    fn stream_pages_inner(
        &self,
        context: Option<&HashMap<String, Value>>,
    ) -> Pin<Box<dyn Stream<Item = Result<faucet_core::StreamPage, FaucetError>> + Send + '_>> {
        // Clone the context into an owned map so it can live inside the
        // `async_stream` generator without borrowing from the caller.
        let owned_context: Option<HashMap<String, Value>> = context.cloned();

        Box::pin(async_stream::try_stream! {
            // Async-job lifecycle (#514): submit → poll → fetch replaces the
            // normal single-GET + pagination flow and yields one result page.
            if self.config.async_job.is_some() {
                let records = self.run_async_job().await?;
                yield faucet_core::StreamPage { records, bookmark: None };
                return;
            }

            // Resolve the effective start-bookmark once at the top of the stream.
            // A runtime override (applied via `Source::apply_start_bookmark` —
            // typically by the pipeline reading from a `StateStore`) takes
            // precedence over the static config value.
            let effective_start: Option<Value> = {
                let guard = self.runtime_start.lock().await;
                guard
                    .clone()
                    .or_else(|| self.config.start_replication_value.clone())
            };

            // H13 (audit #146): combining `max_pages` with incremental
            // replication only makes safe forward progress when the API returns
            // rows ordered ascending by the replication key. On truncation we
            // advance the bookmark to the max key seen so far (so the next run
            // resumes past it — without this the stream would re-read the same
            // first `max_pages` window forever and never progress); but if the
            // feed is unordered, unfetched later pages may hold lower keys that
            // resuming past `running_max` would then drop. Warn loudly so the
            // requirement is explicit rather than a silent data-loss edge.
            if self.config.max_pages.is_some()
                && self.config.replication_method == ReplicationMethod::Incremental
                && self.config.replication_key.is_some()
            {
                tracing::warn!(
                    "max_pages combined with incremental replication assumes the API returns rows \
                     ordered ascending by the replication key; an unordered feed can drop unfetched \
                     lower-key records on resume. Ensure ordering, or remove max_pages for a full \
                     incremental sweep."
                );
            }

            // #527: build the pass plan. Without a `window:` block this is a
            // single "unbounded" pass with the classic record-derived bookmark.
            // With one, each rolling `[start, end)` window is its own pass whose
            // bookmark is the window's end boundary — so a mid-sweep crash resumes
            // from the last completed window (per-window durability).
            let windowed = self.config.window.is_some();
            let passes: Vec<Option<faucet_core::Window>> = if let Some(win) = &self.config.window {
                let start_val = effective_start.clone().ok_or_else(|| {
                    FaucetError::Config(
                        "rest: `window` slicing requires a start bookmark (from a `state:` store) \
                         or `start_replication_value` to anchor the first window".into(),
                    )
                })?;
                let start_instant = faucet_core::parse_instant(&start_val)?;
                let now = self.now_override.unwrap_or_else(chrono::Utc::now);
                let step = win.step_duration()?;
                let lookback = win.lookback_duration()?;
                let (windows, truncated) =
                    faucet_core::enumerate_windows(start_instant, now, step, lookback, win.max_windows);
                if truncated {
                    tracing::warn!(
                        max_windows = win.max_windows,
                        "window slicing hit `max_windows`; this run's sweep is truncated — the next \
                         run resumes from the last completed window"
                    );
                }
                if windows.is_empty() {
                    tracing::debug!(
                        "window slicing: the bookmark is at or ahead of now; nothing to fetch"
                    );
                }
                windows.into_iter().map(Some).collect()
            } else {
                vec![None]
            };

            for pass in passes {
                // Set the window bounds applied to every request in this pass
                // (an unbounded pass leaves `window_binds` empty).
                // `execute_request_once` reads `self.window_binds`.
                if let Some(w) = &pass {
                    let win = self
                        .config
                        .window
                        .as_ref()
                        .expect("a window pass implies a `window:` block");
                    let lower = (win.lower.into, win.lower.name.clone(), win.render_lower(w));
                    let upper_rendered = win.render_upper(w)?;
                    let upper = (win.upper.into, win.upper.name.clone(), upper_rendered);
                    *self.window_binds.lock().await = vec![lower, upper];
                }

                // The bookmark this pass persists on its final page: the window's
                // end (a half-open boundary, so resume neither gaps nor overlaps)
                // for a windowed pass, or the record-derived running max for the
                // classic unbounded pass.
                let window_bookmark: Option<Value> =
                    pass.as_ref().map(|w| Value::String(w.end.to_rfc3339()));

                let mut state = PaginationState::default();
                // #547: on resume, seed the stored cursor into the first request
                // (query param for `Cursor`, body field for `CursorInBody`).
                if self.config.persist_cursor
                    && let Some(seed) = effective_start.as_ref()
                {
                    state.next_token =
                        Some(crate::pagination::value_to_param_string(seed));
                }
                let mut pages_fetched = 0usize;
                let mut running_max: Option<Value> = effective_start.clone();
                // #547: the terminal cursor to persist as this run's bookmark.
                let mut running_cursor: Option<Value> = effective_start.clone();
                let mut bookmark_emitted = false;

                loop {
                    if let Some(max) = self.config.max_pages
                        && pages_fetched >= max
                    {
                        tracing::warn!("max pages ({max}) reached");
                        break;
                    }

                    let mut params = self.config.query_params.clone();
                    self.config.pagination.apply_params(&mut params, &state);

                    let url_override = match &self.config.pagination {
                        PaginationStyle::LinkHeader | PaginationStyle::NextLinkInBody { .. } => {
                            state.next_link.clone()
                        }
                        _ => None,
                    };

                    // Body-carrying pagination (CursorInBody / OffsetInBody /
                    // RecordFieldCursor into:body): fields injected into the
                    // request JSON body for this page.
                    let body_params = self.config.pagination.body_params(&state);

                    let params_clone = params.clone();
                    let ctx_ref = owned_context.as_ref();
                    let is_first_page = pages_fetched == 0;
                    let (body, resp_headers) = retry::execute_with_retry(
                        // The REST runner takes retries-after-first; the policy holds
                        // total attempts. Feed both knobs from the resolved policy so
                        // an injected `resilience:` policy (when legacy fields are
                        // untouched) governs the retry budget + base backoff while the
                        // runner keeps its 429 / `Retry-After` handling.
                        self.retry_policy.max_attempts.saturating_sub(1),
                        self.retry_policy.base,
                        || {
                            self.execute_request(
                                &params_clone,
                                url_override.as_deref(),
                                ctx_ref,
                                is_first_page,
                                &body_params,
                            )
                        },
                    )
                    .await?;

                    let raw_records = self.extract_page(&body)?;
                    let raw_count = raw_records.len();

                    // #547: track the terminal cursor to persist as the bookmark.
                    if self.config.persist_cursor
                        && let Some(path) = self.config.pagination.cursor_path()
                        && let Some(cursor) = jsonpath_first_value(&body, path)
                    {
                        match &cursor {
                            Value::Null => {}
                            Value::String(s) if s.is_empty() => {}
                            _ => running_cursor = Some(cursor),
                        }
                    }

                    // Client-side incremental filter. Skipped for windowed passes:
                    // the server already bounds each window, and filtering by the
                    // overall start would drop `lookback` rows that fall before it.
                    let records = if !windowed
                        && self.config.replication_method == ReplicationMethod::Incremental
                    {
                        if let (Some(key), Some(start)) =
                            (&self.config.replication_key, effective_start.as_ref())
                        {
                            filter_incremental(raw_records, key, start)
                        } else {
                            raw_records
                        }
                    } else {
                        raw_records
                    };

                    // Track the running max replication value across pages so the
                    // final page of an unbounded pass can carry the consolidated
                    // bookmark. When the replication bind declares `advance_from`,
                    // the next bookmark is read from that JSONPath in the response
                    // body (#513); otherwise it is `max(record[replication_key])`.
                    // Windowed passes ignore this — their bookmark is the window end.
                    if !windowed
                        && self.config.replication_method == ReplicationMethod::Incremental
                    {
                        let page_max: Option<Value> = match self
                            .config
                            .replication_bind
                            .as_ref()
                            .and_then(|b| b.advance_from.as_deref())
                        {
                            Some(path) => faucet_core::util::extract_records(&body, Some(path))
                                .ok()
                                .and_then(|vs| vs.into_iter().next()),
                            None => self
                                .config
                                .replication_key
                                .as_deref()
                                .and_then(|key| max_replication_value(&records, key).cloned()),
                        };
                        if let Some(page_max) = page_max {
                            running_max = Some(match running_max.take() {
                                Some(prev) => max_value(prev, page_max),
                                None => page_max,
                            });
                        }
                    }

                    // #554: derive this page's keyset cursor (max/min of the
                    // configured field) so the next request can page by it. A
                    // no-op for every non-RecordFieldCursor style.
                    self.config
                        .pagination
                        .update_record_cursor(&records, &mut state);

                    // Advance pagination state to learn whether there is a next
                    // page BEFORE yielding the current one. This way the bookmark
                    // is only attached to pages where `has_next == false`, and we
                    // never pre-fetch the next page just to classify the current
                    // one as "final" (which would prevent early exit in callers
                    // such as `fetch_partition` with `max_records`).
                    let has_next = self
                        .config
                        .pagination
                        .advance(&body, &resp_headers, &mut state, raw_count)?;
                    pages_fetched += 1;

                    if has_next {
                        // Intermediate page — yield without bookmark so the
                        // pipeline does not persist a partial checkpoint.
                        yield faucet_core::StreamPage { records, bookmark: None };
                    } else if state.current_page_is_duplicate {
                        // The content-stagnation guard flagged this page as a
                        // duplicate of the previous one — DROP it (do not emit the
                        // repeated records to the sink) and stop. The trailing
                        // bookmark checkpoint below still fires (#321 L1).
                        break;
                    } else {
                        // Final page of this pass — attach the pass bookmark.
                        let bookmark = if self.config.persist_cursor {
                            running_cursor.clone()
                        } else if windowed {
                            window_bookmark.clone()
                        } else {
                            running_max.clone()
                        };
                        bookmark_emitted = bookmark.is_some();
                        yield faucet_core::StreamPage { records, bookmark };
                        break;
                    }

                    if let Some(delay) = self.config.request_delay {
                        tokio::time::sleep(delay).await;
                    }
                }

                // Trailing checkpoint: if the pass loop exited without carrying the
                // bookmark on a real page (max_pages truncation, or a duplicate-page
                // stop), emit one empty page carrying the pass bookmark so progress
                // still persists and the next run resumes from here. (Safe forward
                // progress under max_pages assumes ascending order by the
                // replication key — see the warning emitted above, audit #146 H13.)
                let pass_bookmark = if self.config.persist_cursor {
                    running_cursor.clone()
                } else if windowed {
                    window_bookmark.clone()
                } else {
                    running_max.clone()
                };
                if !bookmark_emitted && pass_bookmark.is_some() {
                    yield faucet_core::StreamPage {
                        records: Vec::new(),
                        bookmark: pass_bookmark,
                    };
                }
            }

            // Clear the window bounds so a reused source instance starts clean.
            if windowed {
                self.window_binds.lock().await.clear();
            }
        })
    }

    /// Run the full pagination loop for a single partition context.
    ///
    /// `max_records`: when `Some(n)`, stop collecting after `n` records
    /// (used for schema sampling).
    async fn fetch_partition(
        &self,
        context: Option<&HashMap<String, Value>>,
        max_records: Option<usize>,
    ) -> Result<Vec<Value>, FaucetError> {
        let mut all_records = Vec::new();
        let mut pages_fetched = 0usize;
        let mut pages = self.stream_pages_inner(context);

        // Poll the stream without requiring StreamExt (avoids extra dependency).
        loop {
            let page = std::future::poll_fn(|cx: &mut std::task::Context<'_>| {
                pages.as_mut().poll_next(cx)
            })
            .await;

            match page {
                Some(Ok(page)) => {
                    pages_fetched += 1;
                    let records = page.records;
                    match max_records {
                        Some(limit) => {
                            let remaining = limit.saturating_sub(all_records.len());
                            all_records.extend(records.into_iter().take(remaining));
                            if all_records.len() >= limit {
                                break;
                            }
                        }
                        None => all_records.extend(records),
                    }
                }
                Some(Err(e)) => return Err(e),
                None => break,
            }
        }

        tracing::info!(
            stream = self.config.name.as_deref().unwrap_or("(unnamed)"),
            records = all_records.len(),
            pages = pages_fetched,
            "fetch complete"
        );
        Ok(all_records)
    }

    /// Execute a request, transparently refreshing an inline OAuth2 /
    /// TokenEndpoint token once on a 401.
    ///
    /// The cached token's validity is tracked purely by the server-reported
    /// `expires_in` (and a token with no `expires_in` is cached as valid
    /// forever), so a *server-side* expiry surfaces only as a 401 on a real
    /// request. The documented contract is "valid until a 401 forces a
    /// refresh" — so on a 401 with an inline cached token we invalidate the
    /// cache and retry exactly once with a freshly-fetched token (F57). Shared
    /// auth providers manage their own refresh and are not retried here.
    async fn execute_request(
        &self,
        params: &HashMap<String, String>,
        url_override: Option<&str>,
        path_context: Option<&HashMap<String, Value>>,
        is_first_page: bool,
        body_params: &[(String, Value)],
    ) -> Result<(Value, HeaderMap), FaucetError> {
        match self
            .execute_request_once(
                params,
                url_override,
                path_context,
                is_first_page,
                body_params,
            )
            .await
        {
            Err(FaucetError::HttpStatus { status: 401, .. }) if self.uses_inline_cached_token() => {
                tracing::warn!(
                    "401 Unauthorized with a cached inline OAuth2/TokenEndpoint token; \
                     invalidating the token cache and retrying once with a fresh token"
                );
                self.invalidate_inline_token_cache().await;
                self.execute_request_once(
                    params,
                    url_override,
                    path_context,
                    is_first_page,
                    body_params,
                )
                .await
            }
            // #511: a shared provider (e.g. a multi-step flow) whose session
            // expired mid-run — re-auth on a status it declared in `reauth_on`
            // and retry once.
            Err(FaucetError::HttpStatus { status, .. }) if self.provider_wants_reauth(status) => {
                if let Some(provider) = &self.auth_provider {
                    tracing::warn!(
                        status,
                        "shared auth provider requested re-auth on this status; \
                         re-authenticating and retrying once"
                    );
                    let _ = provider.invalidate(&Credential::Token(String::new())).await;
                }
                self.execute_request_once(
                    params,
                    url_override,
                    path_context,
                    is_first_page,
                    body_params,
                )
                .await
            }
            other => other,
        }
    }

    /// `true` when a shared provider declared `status` in its `reauth_statuses`.
    fn provider_wants_reauth(&self, status: u16) -> bool {
        self.auth_provider
            .as_ref()
            .is_some_and(|p| p.reauth_statuses().contains(&status))
    }

    /// `true` when this source resolves its bearer token from one of the inline
    /// time-cached auth modes (no shared provider) — the only case where a 401
    /// should trigger a cache invalidation + retry (F57).
    fn uses_inline_cached_token(&self) -> bool {
        self.auth_provider.is_none()
            && matches!(
                self.config.auth,
                AuthSpec::Inline(Auth::OAuth2 { .. })
                    | AuthSpec::Inline(Auth::TokenEndpoint { .. })
            )
    }

    /// Invalidate whichever inline token cache backs the current auth mode, so
    /// the next request fetches a fresh token (F57).
    async fn invalidate_inline_token_cache(&self) {
        match &self.config.auth {
            AuthSpec::Inline(Auth::OAuth2 { .. }) => self.token_cache.invalidate().await,
            AuthSpec::Inline(Auth::TokenEndpoint { .. }) => {
                self.token_endpoint_cache.invalidate().await
            }
            _ => {}
        }
    }

    /// Resolve the server-side push-down binding for this run:
    /// `(target, name, rendered-value)`. Returns `None` when no `replication_bind`
    /// is configured or there is no bookmark yet (first run — a full pull).
    async fn resolved_bind(&self) -> Result<Option<(BindTarget, String, String)>, FaucetError> {
        let Some(bind) = &self.config.replication_bind else {
            return Ok(None);
        };
        let bookmark = {
            let guard = self.runtime_start.lock().await;
            guard.clone()
        }
        .or_else(|| self.config.start_replication_value.clone());
        match bookmark {
            Some(bm) => Ok(Some((bind.into, bind.name.clone(), bind.render(&bm)?))),
            None => Ok(None),
        }
    }

    /// Build + send one job-lifecycle request (auth via `metadata_headers`,
    /// plus the connector's static headers and the request's own headers/query/
    /// json). Returns the raw response bytes; errors on non-2xx.
    async fn job_request_bytes(
        &self,
        method: &str,
        url: &str,
        headers: &HashMap<String, String>,
        query: &HashMap<String, String>,
        json: Option<&Value>,
    ) -> Result<(Vec<u8>, HeaderMap), FaucetError> {
        let m = reqwest::Method::from_bytes(method.to_uppercase().as_bytes()).map_err(|_| {
            FaucetError::Config(format!("async_job: invalid HTTP method '{method}'"))
        })?;
        // Precedence: static config headers (base) < auth < this request's own
        // headers — so an auth header always wins over a same-named config one.
        let mut hdrs = self.static_headers.clone();
        for (k, v) in self.metadata_headers(url).await?.iter() {
            hdrs.insert(k.clone(), v.clone());
        }
        for (k, v) in headers {
            insert_header(&mut hdrs, k, v)?;
        }
        let mut req = self.client.request(m, url).headers(hdrs);
        if !query.is_empty() {
            let pairs: Vec<(&str, &str)> = query
                .iter()
                .map(|(k, v)| (k.as_str(), v.as_str()))
                .collect();
            req = req.query(&pairs);
        }
        if let Some(j) = json {
            req = req.json(j);
        }
        let resp = req
            .send()
            .await
            .map_err(|e| FaucetError::Source(format!("async_job: request to {url} failed: {e}")))?;
        let status = resp.status();
        if !status.is_success() {
            return Err(FaucetError::HttpStatus {
                status: status.as_u16(),
                url: url.to_string(),
                body: format!("async_job: {url} returned HTTP {}", status.as_u16()),
            });
        }
        let resp_headers = resp.headers().clone();
        Ok((resp.bytes().await?.to_vec(), resp_headers))
    }

    async fn job_request_json(
        &self,
        method: &str,
        url: &str,
        headers: &HashMap<String, String>,
        query: &HashMap<String, String>,
        json: Option<&Value>,
    ) -> Result<Value, FaucetError> {
        let (bytes, _headers) = self
            .job_request_bytes(method, url, headers, query, json)
            .await?;
        serde_json::from_slice(&bytes)
            .map_err(|e| FaucetError::Source(format!("async_job: {url} returned non-JSON: {e}")))
    }

    /// Run the submit → poll → fetch job lifecycle (#514) and return the
    /// decoded result records.
    async fn run_async_job(&self) -> Result<Vec<Value>, FaucetError> {
        use crate::async_job::{JobOutcome, resolve_url, substitute_job_id};
        let job = self
            .config
            .async_job
            .as_ref()
            .expect("run_async_job called with async_job set");
        let base = &self.config.base_url;

        // 1) Submit → capture the job id.
        let submit_url = resolve_url(base, job.submit.url.as_deref().unwrap_or_default());
        let submit_body = self
            .job_request_json(
                &job.submit.method,
                &submit_url,
                &job.submit.headers,
                &job.submit.query,
                job.submit.json.as_ref(),
            )
            .await?;
        let job_id = jsonpath_first_string(&submit_body, &job.job_id).ok_or_else(|| {
            FaucetError::Source(format!(
                "async_job: submit response had no job id at '{}'",
                job.job_id
            ))
        })?;

        // 2) Poll until a terminal state (with interval + timeout).
        let poll_url = resolve_url(base, &substitute_job_id(&job.poll.url, &job_id));
        let deadline =
            tokio::time::Instant::now() + std::time::Duration::from_secs(job.poll.timeout_secs);
        // Retain the last poll response so `fetch.url_from` (#543) can source the
        // download URL from the terminal (success) poll body.
        let last_poll_body: Value = loop {
            let body = self
                .job_request_json(
                    &job.poll.method,
                    &poll_url,
                    &job.poll.headers,
                    &job.poll.query,
                    None,
                )
                .await?;
            let status = jsonpath_first_string(&body, &job.status.path).unwrap_or_default();
            match job.status.classify(&status) {
                JobOutcome::Success => break body,
                JobOutcome::Failure => {
                    return Err(FaucetError::Source(format!(
                        "async_job: job failed with status '{status}'"
                    )));
                }
                JobOutcome::Pending => {
                    if tokio::time::Instant::now() >= deadline {
                        return Err(FaucetError::Source(format!(
                            "async_job: polling timed out after {}s (last status '{status}')",
                            job.poll.timeout_secs
                        )));
                    }
                    tokio::time::sleep(std::time::Duration::from_secs(job.poll.interval_secs))
                        .await;
                }
            }
        };

        // 3) Resolve the fetch URL (#543): from the poll body via `url_from`, or
        // by rendering the templated `url`. Exactly one is set (validated).
        let fetch_url = match (&job.fetch.url_from, &job.fetch.url) {
            (Some(path), _) => {
                let resolved = jsonpath_first_string(&last_poll_body, path).ok_or_else(|| {
                    FaucetError::Source(format!(
                        "async_job: fetch.url_from '{path}' matched no string in the poll response"
                    ))
                })?;
                resolve_url(base, &resolved)
            }
            (None, Some(url)) => resolve_url(base, &substitute_job_id(url, &job_id)),
            (None, None) => {
                return Err(FaucetError::Config(
                    "async_job: `fetch` requires exactly one of `url` or `url_from`".into(),
                ));
            }
        };

        // 4) Fetch the result and decode it — looping across locator-paged result
        // sets (#557) when a `locator_header` / `locator_body` is configured.
        // Without a locator this runs exactly once (the classic single fetch).
        let mut all_records = Vec::new();
        let mut locator: Option<String> = None;
        loop {
            // Send the locator (when we have one) as the configured query param.
            let mut query = job.fetch.query.clone();
            if let (Some(loc), Some(param)) = (&locator, &job.fetch.locator_param) {
                query.insert(param.clone(), loc.clone());
            }
            let (bytes, resp_headers) = self
                .job_request_bytes(
                    &job.fetch.method,
                    &fetch_url,
                    &job.fetch.headers,
                    &query,
                    job.fetch.json.as_ref(),
                )
                .await?;
            let (records, body_value) = self.parse_fetch_page(&bytes, job).await?;
            all_records.extend(records);

            // Determine the next locator from the header or the body; stop when
            // it is absent, empty, `"null"`, or repeats (loop guard).
            let next = next_locator(&resp_headers, body_value.as_ref(), job);
            match next {
                Some(loc) if locator.as_deref() != Some(loc.as_str()) => {
                    locator = Some(loc);
                }
                _ => break,
            }
        }
        Ok(all_records)
    }

    /// Parse one async-job fetch page into records, returning the parsed JSON
    /// body too (for `locator_body` extraction) when the result is JSON. Mirrors
    /// the single-fetch parsing: a `decode:` pipeline wins, else `response_format`
    /// (JSON honouring `fetch.records_path` or the source `records_path`).
    async fn parse_fetch_page(
        &self,
        bytes: &[u8],
        job: &crate::async_job::AsyncJobConfig,
    ) -> Result<(Vec<Value>, Option<Value>), FaucetError> {
        if !self.config.decode.is_empty() {
            let records = crate::decode::run_decode(bytes, &self.config.decode).await?;
            return Ok((records, None));
        }
        match self.config.response_format {
            crate::config::ResponseFormat::Json => {
                let v: Value = serde_json::from_slice(bytes).map_err(|e| {
                    FaucetError::Source(format!("async_job: result is not JSON: {e}"))
                })?;
                let records = match job.fetch.records_path.as_deref() {
                    Some(rp) => extract::extract_records(&v, Some(rp))?,
                    None => self.extract_page(&v)?,
                };
                Ok((records, Some(v)))
            }
            crate::config::ResponseFormat::Csv => {
                let records = crate::format::parse_csv(
                    bytes,
                    self.config.csv_delimiter,
                    self.config.csv_has_headers,
                )
                .await?;
                Ok((records, None))
            }
            crate::config::ResponseFormat::Excel => {
                let records = crate::format::parse_excel(
                    bytes,
                    self.config.excel_sheet.as_deref(),
                    self.config.excel_header_row,
                )?;
                Ok((records, None))
            }
        }
    }

    /// Resolve auth headers for a non-paginated preflight request (OData
    /// `$metadata`). Applies a flow provider's header/cookie placements or its
    /// credential; else the inline auth (bearer via cache for OAuth2/token
    /// endpoint). Query/body placements and `ApiKeyQuery` are not applied here.
    async fn metadata_headers(&self, url: &str) -> Result<HeaderMap, FaucetError> {
        let mut headers = HeaderMap::new();
        if let Some(provider) = &self.auth_provider {
            let ra = provider
                .request_auth("GET", url, &std::collections::BTreeMap::new())
                .await?;
            if ra.is_empty() {
                credential_to_auth(provider.credential().await?).apply(&mut headers)?;
            } else {
                for p in ra.placements {
                    match p {
                        CredentialPlacement::Header { name, value } => {
                            insert_header(&mut headers, &name, &value)?
                        }
                        CredentialPlacement::Cookie { name, value } => {
                            insert_header(&mut headers, "Cookie", &format!("{name}={value}"))?
                        }
                        _ => {}
                    }
                }
            }
        } else {
            match &self.config.auth {
                AuthSpec::Inline(Auth::OAuth2 {
                    token_url,
                    client_id,
                    client_secret,
                    scopes,
                    expiry_ratio,
                }) => {
                    let token = self
                        .token_cache
                        .get_or_refresh(
                            &self.client,
                            token_url,
                            client_id,
                            client_secret,
                            scopes,
                            *expiry_ratio,
                        )
                        .await?;
                    Auth::Bearer { token }.apply(&mut headers)?;
                }
                AuthSpec::Inline(Auth::TokenEndpoint {
                    url: token_url,
                    method: token_method,
                    headers: token_headers,
                    body: token_body,
                    token_path,
                    expiry_path,
                    expiry_ratio,
                    response_validator,
                }) => {
                    let token = self
                        .token_endpoint_cache
                        .get_or_refresh(
                            &self.client,
                            token_url,
                            token_method,
                            token_headers,
                            token_body.as_ref(),
                            token_path,
                            expiry_path.as_deref(),
                            *expiry_ratio,
                            response_validator.as_ref(),
                        )
                        .await?;
                    Auth::Bearer { token }.apply(&mut headers)?;
                }
                AuthSpec::Inline(other) => other.apply(&mut headers)?,
                AuthSpec::Reference(_) => {}
            }
        }
        Ok(headers)
    }

    /// Execute a single HTTP request and return the response body and headers.
    ///
    /// - When `url_override` is `Some`, that full URL is used and query params
    ///   are **not** appended (Link header pagination encodes them in the URL).
    /// - When `path_context` is `Some`, `{key}` placeholders in `config.path`
    ///   are substituted with values from the context map (partition support).
    async fn execute_request_once(
        &self,
        params: &HashMap<String, String>,
        url_override: Option<&str>,
        path_context: Option<&HashMap<String, Value>>,
        is_first_page: bool,
        body_params: &[(String, Value)],
    ) -> Result<(Value, HeaderMap), FaucetError> {
        let use_override = url_override.is_some();

        // #513 server-side push-down + #527 window slicing: the outgoing request
        // carries the bookmark binding (0 or 1) plus the current window's rendered
        // lower/upper bounds (0 or 2). They apply at the same four placement sites.
        let mut binds: Vec<(BindTarget, String, String)> = Vec::new();
        if let Some(b) = self.resolved_bind().await? {
            binds.push(b);
        }
        binds.extend(self.window_binds.lock().await.iter().cloned());

        let query_btree: std::collections::BTreeMap<String, String> =
            params.iter().map(|(k, v)| (k.clone(), v.clone())).collect();

        // #511 rich per-request auth: a flow provider may place credentials
        // across header/query/cookie/body and override the base-URL for this
        // session. When it contributes anything, it supersedes the plain
        // credential()/sign_request() path below.
        let mut base_url = self.config.base_url.clone();
        let mut ra_headers: Vec<(String, String)> = Vec::new();
        let mut ra_query: Vec<(String, String)> = Vec::new();
        let mut ra_cookies: Vec<(String, String)> = Vec::new();
        let mut ra_body: Vec<(String, String)> = Vec::new();
        let mut captured: std::collections::BTreeMap<String, String> =
            std::collections::BTreeMap::new();
        let mut used_request_auth = false;
        if let Some(provider) = &self.auth_provider {
            let ra = provider
                .request_auth(self.config.method.as_str(), &base_url, &query_btree)
                .await?;
            if !ra.is_empty() {
                used_request_auth = true;
                if let Some(b) = ra.base_url {
                    base_url = b;
                }
                captured = ra.captured;
                for p in ra.placements {
                    match p {
                        CredentialPlacement::Header { name, value } => {
                            ra_headers.push((name, value))
                        }
                        CredentialPlacement::Query { name, value } => ra_query.push((name, value)),
                        CredentialPlacement::Cookie { name, value } => {
                            ra_cookies.push((name, value))
                        }
                        CredentialPlacement::BodyField { name, value } => {
                            ra_body.push((name, value))
                        }
                        _ => {}
                    }
                }
            }
        }

        // Build the URL (honouring any dynamic base-URL) and apply a `path`-target
        // push-down binding.
        let mut url = match url_override {
            Some(u) => u.to_string(),
            None => {
                let path = match path_context {
                    Some(ctx) => faucet_core::util::substitute_context(&self.config.path, ctx),
                    None => self.config.path.clone(),
                };
                format!("{}/{}", base_url, path.trim_start_matches('/'))
            }
        };
        for (target, name, rendered) in &binds {
            if *target == BindTarget::Path {
                url = url.replace(&format!("{{{name}}}"), rendered);
            }
        }
        // #567: substitute flow-captured `${name}` values into the URL (a
        // captured session id in the path, say). No-op when nothing was captured.
        url = substitute_captured(&url, &captured);

        // Resolve inline / signed credentials — unless a flow provider already
        // supplied the request auth. A shared provider (from `auth: { ref }` or
        // a library caller) takes precedence over inline; inline OAuth2 /
        // TokenEndpoint resolve to a Bearer token via the per-source cache.
        let resolved_auth: Option<Auth> = if used_request_auth {
            None
        } else if let Some(provider) = &self.auth_provider {
            // A per-request signer (OAuth1, #496) signs this exact method + URL +
            // query; every other provider returns `None` here and we apply its
            // reusable credential.
            let cred = match provider
                .sign_request(self.config.method.as_str(), &url, &query_btree)
                .await?
            {
                Some(cred) => cred,
                None => provider.credential().await?,
            };
            Some(credential_to_auth(cred))
        } else {
            match &self.config.auth {
                AuthSpec::Inline(Auth::OAuth2 {
                    token_url,
                    client_id,
                    client_secret,
                    scopes,
                    expiry_ratio,
                }) => {
                    let token = self
                        .token_cache
                        .get_or_refresh(
                            &self.client,
                            token_url,
                            client_id,
                            client_secret,
                            scopes,
                            *expiry_ratio,
                        )
                        .await?;
                    Some(Auth::Bearer { token })
                }
                AuthSpec::Inline(Auth::TokenEndpoint {
                    url: token_url,
                    method: token_method,
                    headers: token_headers,
                    body: token_body,
                    token_path,
                    expiry_path,
                    expiry_ratio,
                    response_validator,
                }) => {
                    let token = self
                        .token_endpoint_cache
                        .get_or_refresh(
                            &self.client,
                            token_url,
                            token_method,
                            token_headers,
                            token_body.as_ref(),
                            token_path,
                            expiry_path.as_deref(),
                            *expiry_ratio,
                            response_validator.as_ref(),
                        )
                        .await?;
                    Some(Auth::Bearer { token })
                }
                AuthSpec::Inline(other) => Some(other.clone()),
                AuthSpec::Reference(r) => {
                    return Err(FaucetError::Auth(format!(
                        "auth references provider '{}' but no provider was supplied; \
                         set one via the CLI `auth:` catalog or `with_auth_provider`",
                        r.name
                    )));
                }
            }
        };

        // Static config headers form the base; auth (inline or provider) is
        // applied on top so an auth header of the same name wins (#539). A
        // flow-captured `${name}` in a header value is substituted per request
        // (#567); the map is empty (and this a plain clone) for non-flow auth.
        let mut headers = if captured.is_empty() {
            self.static_headers.clone()
        } else {
            let mut h = HeaderMap::new();
            for (name, value) in self.static_headers.iter() {
                let sv = substitute_captured(value.to_str().unwrap_or_default(), &captured);
                let hv =
                    reqwest::header::HeaderValue::from_str(&sv).unwrap_or_else(|_| value.clone());
                h.insert(name.clone(), hv);
            }
            h
        };
        if let Some(auth) = &resolved_auth {
            auth.apply(&mut headers)?;
        }
        // #511 header + cookie placements from the flow provider.
        for (name, value) in &ra_headers {
            insert_header(&mut headers, name, value)?;
        }
        if !ra_cookies.is_empty() {
            let cookie = ra_cookies
                .iter()
                .map(|(k, v)| format!("{k}={v}"))
                .collect::<Vec<_>>()
                .join("; ");
            insert_header(&mut headers, "Cookie", &cookie)?;
        }
        // #513/#527 header-target bindings.
        for (target, name, rendered) in &binds {
            if *target == BindTarget::Header {
                insert_header(&mut headers, name, rendered)?;
            }
        }

        let mut req = self
            .client
            .request(self.config.method.clone(), &url)
            .headers(headers);

        if !use_override {
            // When parent context is available, substitute {placeholders} in
            // query param values so child sources can be parameterised.
            if let Some(ctx) = path_context {
                let substituted: HashMap<String, String> = params
                    .iter()
                    .map(|(k, v)| (k.clone(), faucet_core::util::substitute_context(v, ctx)))
                    .collect();
                req = req.query(&substituted.iter().collect::<Vec<_>>());
            } else {
                req = req.query(params);
            }
            // #536: repeated / array-valued query params, rendered as repeated
            // keys (`?k=a&k=b`). reqwest's `.query()` appends, so this composes
            // with the scalar params above.
            if !self.config.query_params_multi.is_empty() {
                let pairs: Vec<(String, String)> = self
                    .config
                    .query_params_multi
                    .iter()
                    .flat_map(|(k, vals)| {
                        vals.iter().map(move |v| {
                            let rendered = match path_context {
                                Some(ctx) => faucet_core::util::substitute_context(v, ctx),
                                None => v.clone(),
                            };
                            (k.clone(), rendered)
                        })
                    })
                    .collect();
                req = req.query(
                    &pairs
                        .iter()
                        .map(|(k, v)| (k.as_str(), v.as_str()))
                        .collect::<Vec<_>>(),
                );
            }
        }
        // #511 query placements from the flow provider.
        if !ra_query.is_empty() {
            let pairs: Vec<(&str, &str)> = ra_query
                .iter()
                .map(|(k, v)| (k.as_str(), v.as_str()))
                .collect();
            req = req.query(&pairs);
        }
        // #513/#527 query-target bindings.
        for (target, name, rendered) in &binds {
            if *target == BindTarget::Query {
                req = req.query(&[(name.as_str(), rendered.as_str())]);
            }
        }

        // ApiKeyQuery: inject the API key as a query parameter.
        if let AuthSpec::Inline(Auth::ApiKeyQuery { param, value }) = &self.config.auth {
            req = req.query(&[(param.as_str(), value.as_str())]);
        }

        // Build the request JSON body, if any. Substitute context into body
        // string values when available. Use the JSON-safe variant:
        // `substitute_context` does NOT escape the value, so a context value
        // carrying a JSON metacharacter (`"`, `\`, newline) corrupts the
        // serialized body — the old `unwrap_or(Value::String(..))` fallback then
        // silently coerced the whole object into a bare string and POSTed garbage
        // (audit #321 H7). `substitute_context_json` JSON-escapes string values;
        // an un-parseable result is now a hard error rather than a silently-wrong
        // payload.
        let mut body_value: Option<Value> = match &self.config.body {
            Some(body) => match path_context {
                Some(ctx) => {
                    let body_str = body.to_string();
                    let substituted = faucet_core::util::substitute_context_json(&body_str, ctx);
                    let substituted_value: Value =
                        serde_json::from_str(&substituted).map_err(|e| {
                            FaucetError::Source(format!(
                                "REST source: context substitution produced an invalid JSON body: {e}"
                            ))
                        })?;
                    Some(substituted_value)
                }
                None => Some(body.clone()),
            },
            None => None,
        };
        // Body-carrying pagination (CursorInBody / OffsetInBody / RecordFieldCursor
        // with `into: body`): inject the pagination fields into the request body.
        // If no base body was configured, start from an empty object so the
        // fields still land somewhere.
        if !body_params.is_empty() {
            let obj = body_value.get_or_insert_with(|| Value::Object(serde_json::Map::new()));
            match obj.as_object_mut() {
                Some(map) => {
                    for (field, value) in body_params {
                        map.insert(field.clone(), value.clone());
                    }
                }
                None => {
                    return Err(FaucetError::Source(
                        "REST source: body-carrying pagination requires a JSON object request \
                         body to inject the pagination fields into"
                            .into(),
                    ));
                }
            }
        }
        // #511 body-field placements + #513/#527 body-target bindings.
        let has_body_bind = binds.iter().any(|(t, _, _)| *t == BindTarget::Body);
        if !ra_body.is_empty() || has_body_bind {
            let obj = body_value.get_or_insert_with(|| Value::Object(serde_json::Map::new()));
            match obj.as_object_mut() {
                Some(map) => {
                    for (name, value) in &ra_body {
                        map.insert(name.clone(), Value::String(value.clone()));
                    }
                    for (target, name, rendered) in &binds {
                        if *target == BindTarget::Body {
                            map.insert(name.clone(), Value::String(rendered.clone()));
                        }
                    }
                }
                None => {
                    return Err(FaucetError::Source(
                        "REST source: a body-target auth/replication binding requires a JSON \
                         object request body"
                            .into(),
                    ));
                }
            }
        }
        if let Some(body) = &body_value {
            req = req.json(body);
        }

        let resp = req.send().await?;
        let status = resp.status();

        // 429 Too Many Requests: honour Retry-After before retrying.
        if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
            let wait = parse_retry_after(resp.headers());
            return Err(FaucetError::RateLimited(wait));
        }

        // Tolerated errors: treat as an empty page ONLY on the first request,
        // where they legitimately mean "this resource is absent/empty". Mid-
        // pagination, an empty page makes every pagination style read "last
        // page" and stop, silently dropping every remaining page as a
        // "successful" run (#78/#7). There we fall through to the real error
        // path: the retry executor retries 5xx, and a persistent error fails
        // loudly instead of truncating the stream.
        if is_first_page && self.config.tolerated_http_errors.contains(&status.as_u16()) {
            tracing::debug!(
                status = status.as_u16(),
                "tolerated HTTP error on first request; treating as empty page"
            );
            return Ok((Value::Array(vec![]), HeaderMap::new()));
        }
        if !is_first_page && self.config.tolerated_http_errors.contains(&status.as_u16()) {
            tracing::warn!(
                status = status.as_u16(),
                "tolerated HTTP error mid-pagination; surfacing as an error to avoid \
                 silently truncating the stream"
            );
        }

        // For non-success responses, capture the body for debugging before
        // returning the error. This gives callers (and logs) the server's
        // error message rather than just a status code.
        if !status.is_success() {
            // Redact any auth secret carried in the query string before it lands
            // in the error (which renders the URL in `Display` → logs). The
            // `api_key_query` value is user-configured, so it is not marked
            // sensitive like a Bearer/Basic header and would otherwise leak on
            // any 4xx/5xx (audit #321 L2).
            let resp_url = redact_error_url(resp.url(), &self.config.auth);
            let body_text = resp.text().await.unwrap_or_default();
            // Truncate very long error bodies to avoid bloating logs/errors.
            let truncated = if body_text.len() > 1024 {
                // Find a safe UTF-8 boundary at or before 1024 bytes.
                let end = body_text.floor_char_boundary(1024);
                format!("{}...(truncated)", &body_text[..end])
            } else {
                body_text
            };
            return Err(FaucetError::HttpStatus {
                status: status.as_u16(),
                url: resp_url,
                body: truncated,
            });
        }

        let resp_headers = resp.headers().clone();

        // A 204 No Content — or any 2xx with an empty / whitespace-only body —
        // carries no JSON to parse. `resp.json()` on such a response yields a
        // non-retriable decode error ("EOF while parsing a value") that aborts
        // the run; treat it as an empty page ("no data") instead (#146 M10). A
        // non-empty body that isn't valid JSON still surfaces as a parse error.
        if status == reqwest::StatusCode::NO_CONTENT {
            return Ok((Value::Array(vec![]), resp_headers));
        }
        let bytes = resp.bytes().await?;
        if bytes.iter().all(u8::is_ascii_whitespace) {
            return Ok((Value::Array(vec![]), resp_headers));
        }
        // A `decode:` pipeline (#515) takes the raw body and produces records
        // directly (extract → base64 → gunzip/unzip → parse). It replaces the
        // `response_format` parsing; `validate()` guarantees pagination is
        // `none`. The records land as an array the downstream
        // (records_path-less) extraction passes straight through.
        if !self.config.decode.is_empty() {
            let records = crate::decode::run_decode(&bytes, &self.config.decode).await?;
            return Ok((Value::Array(records), resp_headers));
        }
        // For file response formats the whole body is a tabular file — parse it
        // into a record array here so the downstream (records_path-less)
        // extraction passes it straight through. `validate()` guarantees
        // pagination is `none`, so a single response is fetched.
        let body: Value = match self.config.response_format {
            crate::config::ResponseFormat::Json => serde_json::from_slice(&bytes)?,
            crate::config::ResponseFormat::Csv => Value::Array(
                crate::format::parse_csv(
                    &bytes,
                    self.config.csv_delimiter,
                    self.config.csv_has_headers,
                )
                .await?,
            ),
            crate::config::ResponseFormat::Excel => Value::Array(crate::format::parse_excel(
                &bytes,
                self.config.excel_sheet.as_deref(),
                self.config.excel_header_row,
            )?),
        };
        Ok((body, resp_headers))
    }
}

/// Render a response URL for an error message with any auth secret in the query
/// string redacted (audit #321 L2). Redacts the user-configured
/// `api_key_query` parameter by name (which `redact_uri_credentials` cannot
/// know), then applies the shared credential/query-secret redaction for the
/// common key names and any URL userinfo.
fn redact_error_url(url: &reqwest::Url, auth: &AuthSpec<Auth>) -> String {
    let mut redacted = url.clone();
    if let AuthSpec::Inline(Auth::ApiKeyQuery { param, .. }) = auth {
        let pairs: Vec<(String, String)> = url
            .query_pairs()
            .map(|(k, v)| {
                if k == param.as_str() {
                    (k.into_owned(), "***".to_string())
                } else {
                    (k.into_owned(), v.into_owned())
                }
            })
            .collect();
        redacted.set_query(None);
        if !pairs.is_empty() {
            let mut qp = redacted.query_pairs_mut();
            for (k, v) in &pairs {
                qp.append_pair(k, v);
            }
        }
    }
    faucet_core::redact_uri_credentials(redacted.as_str())
}

/// Parse the `Retry-After` header. RFC 7231 permits **either** delta-seconds
/// **or** an HTTP-date; we honour both. An HTTP-date in the past yields a zero
/// wait (retry now). Falls back to 60 s only when the header is absent or in
/// neither form.
fn parse_retry_after(headers: &HeaderMap) -> Duration {
    const DEFAULT: Duration = Duration::from_secs(60);
    let Some(raw) = headers
        .get(reqwest::header::RETRY_AFTER)
        .and_then(|v| v.to_str().ok())
        .map(str::trim)
    else {
        return DEFAULT;
    };
    // delta-seconds form.
    if let Ok(secs) = raw.parse::<u64>() {
        return Duration::from_secs(secs);
    }
    // HTTP-date form (IMF-fixdate / RFC 850 / asctime).
    if let Ok(when) = httpdate::parse_http_date(raw) {
        return when
            .duration_since(std::time::SystemTime::now())
            .unwrap_or(Duration::ZERO);
    }
    DEFAULT
}

/// Keep the larger of two bookmark values when consolidating per-partition
/// bookmarks in [`Source::stream_pages`] (#535). Numbers compare numerically,
/// strings lexicographically (the usual timestamp/id bookmark shapes); any
/// other or heterogeneous pair prefers the newer value.
fn value_max(current: Option<Value>, candidate: Value) -> Option<Value> {
    match current {
        None => Some(candidate),
        Some(cur) => {
            let take_candidate = match (&cur, &candidate) {
                (Value::Number(a), Value::Number(b)) => {
                    b.as_f64().unwrap_or(f64::MIN) > a.as_f64().unwrap_or(f64::MIN)
                }
                (Value::String(a), Value::String(b)) => b > a,
                _ => true,
            };
            Some(if take_candidate { candidate } else { cur })
        }
    }
}

#[async_trait]
impl faucet_core::Source for RestStream {
    async fn fetch_with_context(
        &self,
        context: &std::collections::HashMap<String, serde_json::Value>,
    ) -> Result<Vec<Value>, FaucetError> {
        if context.is_empty() {
            // No parent context — use normal fetch_all with partitions
            RestStream::fetch_all(self).await
        } else if self.config.partitions.is_empty() {
            // Parent context, no partitions — use context directly as partition context
            self.fetch_partition(Some(context), None).await
        } else {
            // Both parent context and partitions — merge context into each partition
            let mut all_records = Vec::new();
            for partition in &self.config.partitions {
                let mut merged = context.clone();
                merged.extend(partition.iter().map(|(k, v)| (k.clone(), v.clone())));
                all_records.extend(self.fetch_partition(Some(&merged), None).await?);
            }
            Ok(all_records)
        }
    }

    async fn fetch_with_context_incremental(
        &self,
        context: &std::collections::HashMap<String, serde_json::Value>,
    ) -> Result<(Vec<Value>, Option<Value>), FaucetError> {
        let records = self.fetch_with_context(context).await?;
        let bookmark = self
            .config
            .replication_key
            .as_deref()
            .and_then(|key| faucet_core::replication::max_replication_value(&records, key))
            .cloned();
        Ok((records, bookmark))
    }

    fn connector_name(&self) -> &'static str {
        "rest"
    }

    fn config_schema(&self) -> serde_json::Value {
        serde_json::to_value(faucet_core::schema_for!(RestStreamConfig))
            .expect("schema serialization")
    }

    fn dataset_uri(&self) -> String {
        format!(
            "{}{}",
            faucet_core::redact_uri_credentials(&self.config.base_url),
            self.config.path
        )
    }

    fn state_key(&self) -> Option<String> {
        self.config.state_key.clone()
    }

    fn stream_pages<'a>(
        &'a self,
        context: &'a HashMap<String, Value>,
        _batch_size: usize,
    ) -> Pin<Box<dyn Stream<Item = Result<faucet_core::StreamPage, FaucetError>> + Send + 'a>> {
        // RestStream chunks by upstream-API page boundaries, not by an
        // in-memory `batch_size` knob. The arg is accepted for trait
        // conformance and reserved for a future `page_size` mapping.
        //
        // Partition fan-out (#535): when `partitions` are configured the stream
        // must run once per partition — mirroring `fetch_all` / `fetch_with_context`
        // — or every partition's records are silently dropped under `faucet run`
        // (the pipeline drives this method). Any parent `context` is merged into
        // each partition context, exactly as `fetch_with_context` does.
        if self.config.partitions.is_empty() {
            return self.stream_pages_inner(Some(context));
        }
        let contexts: Vec<HashMap<String, Value>> = self
            .config
            .partitions
            .iter()
            .map(|p| {
                let mut merged = context.clone();
                merged.extend(p.iter().map(|(k, v)| (k.clone(), v.clone())));
                merged
            })
            .collect();
        Box::pin(async_stream::try_stream! {
            // Per-partition streams each emit their own final bookmark; we
            // suppress those and emit a single consolidated (max) bookmark after
            // the last partition, so the persisted state is the global high-water
            // mark rather than whichever partition happened to finish last.
            let mut max_bookmark: Option<Value> = None;
            for ctx in &contexts {
                let mut inner = self.stream_pages_inner(Some(ctx));
                loop {
                    let page = std::future::poll_fn(|cx| inner.as_mut().poll_next(cx)).await;
                    match page {
                        Some(Ok(p)) => {
                            if let Some(bm) = p.bookmark {
                                max_bookmark = value_max(max_bookmark.take(), bm);
                                yield faucet_core::StreamPage { records: p.records, bookmark: None };
                            } else {
                                yield p;
                            }
                        }
                        Some(Err(e)) => Err(e)?,
                        None => break,
                    }
                }
            }
            if max_bookmark.is_some() {
                yield faucet_core::StreamPage { records: Vec::new(), bookmark: max_bookmark };
            }
        })
    }

    async fn apply_start_bookmark(&self, bookmark: Value) -> Result<(), FaucetError> {
        *self.runtime_start.lock().await = Some(bookmark);
        Ok(())
    }

    fn supports_discover(&self) -> bool {
        // OData exposes a machine-readable `$metadata` catalog; a plain REST API
        // has none, so discovery is OData-only.
        self.config.odata.is_some()
    }

    async fn discover(&self) -> Result<Vec<faucet_core::DatasetDescriptor>, FaucetError> {
        if self.config.odata.is_none() {
            return Err(FaucetError::Source(
                "rest: discovery is only supported for OData sources — set an `odata:` block"
                    .into(),
            ));
        }
        let url = format!("{}/$metadata", self.config.base_url.trim_end_matches('/'));
        // Static config headers (#539) form the base; auth is applied on top.
        let mut headers = self.static_headers.clone();
        for (k, v) in self.metadata_headers(&url).await?.iter() {
            headers.insert(k.clone(), v.clone());
        }
        let resp = self
            .client
            .get(&url)
            .headers(headers)
            .send()
            .await
            .map_err(|e| {
                FaucetError::Source(format!("rest: OData $metadata request failed: {e}"))
            })?;
        let status = resp.status();
        if !status.is_success() {
            return Err(FaucetError::Source(format!(
                "rest: OData $metadata returned HTTP {}",
                status.as_u16()
            )));
        }
        let xml = resp.text().await.map_err(|e| {
            FaucetError::Source(format!("rest: reading OData $metadata failed: {e}"))
        })?;
        crate::odata::descriptors_from_edmx(&xml)
    }
}

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

    #[test]
    fn value_max_consolidates_partition_bookmarks() {
        // First value seeds the max.
        assert_eq!(
            value_max(None, json!("2026-01-01")),
            Some(json!("2026-01-01"))
        );
        // Strings compare lexicographically (ISO timestamps sort correctly).
        assert_eq!(
            value_max(Some(json!("2026-01-01")), json!("2026-03-01")),
            Some(json!("2026-03-01"))
        );
        assert_eq!(
            value_max(Some(json!("2026-03-01")), json!("2026-01-01")),
            Some(json!("2026-03-01"))
        );
        // Numbers compare numerically.
        assert_eq!(value_max(Some(json!(5)), json!(10)), Some(json!(10)));
        assert_eq!(value_max(Some(json!(10)), json!(5)), Some(json!(10)));
        // Heterogeneous / other → prefer the latest candidate.
        assert_eq!(value_max(Some(json!("a")), json!(3)), Some(json!(3)));
    }

    #[test]
    fn injected_policy_applies_when_legacy_fields_at_defaults() {
        // Config left at the default max_retries/retry_backoff → injection wins.
        let stream =
            RestStream::new(RestStreamConfig::new("https://api.example.com", "/items")).unwrap();
        let injected = faucet_core::RetryPolicy {
            max_attempts: 9,
            base: Duration::from_secs(7),
            ..faucet_core::RetryPolicy::default()
        };
        let stream = stream.with_retry_policy(injected);
        assert_eq!(stream.retry_policy.max_attempts, 9);
        assert_eq!(stream.retry_policy.base, Duration::from_secs(7));
    }

    #[test]
    fn legacy_fields_take_precedence_over_injected_policy() {
        // User set max_retries explicitly → the injected policy is ignored and
        // the connector's own legacy fields keep governing retries.
        let config = RestStreamConfig::new("https://api.example.com", "/items").max_retries(7);
        let stream = RestStream::new(config).unwrap();
        // Default policy derived from legacy fields: max_attempts = 7 + 1.
        assert_eq!(stream.retry_policy.max_attempts, 8);
        let injected = faucet_core::RetryPolicy {
            max_attempts: 99,
            base: Duration::from_secs(42),
            ..faucet_core::RetryPolicy::default()
        };
        let stream = stream.with_retry_policy(injected);
        // Unchanged: the legacy max_retries(7) still wins.
        assert_eq!(stream.retry_policy.max_attempts, 8);
        assert_eq!(stream.retry_policy.base, DEFAULT_RETRY_BACKOFF);
    }

    #[test]
    fn redact_error_url_hides_api_key_query_param() {
        // #321 L2: a custom `api_key_query` param name is redacted by name.
        let auth = AuthSpec::Inline(Auth::ApiKeyQuery {
            param: "api_token".into(),
            value: "SUPERSECRET".into(),
        });
        let url =
            reqwest::Url::parse("https://api.example.com/v1/items?page=2&api_token=SUPERSECRET")
                .unwrap();
        let redacted = redact_error_url(&url, &auth);
        assert!(
            !redacted.contains("SUPERSECRET"),
            "secret must be gone: {redacted}"
        );
        assert!(redacted.contains("api_token=%2A%2A%2A") || redacted.contains("api_token=***"));
        assert!(
            redacted.contains("page=2"),
            "non-secret param kept: {redacted}"
        );
    }

    #[test]
    fn redact_error_url_without_api_key_query_still_scrubs_common_keys() {
        // Non-ApiKeyQuery auth: the shared redaction still strips common secret
        // query keys and userinfo.
        let auth: AuthSpec<Auth> = AuthSpec::Inline(Auth::None);
        let url = reqwest::Url::parse("https://u:pw@api.example.com/v1/items?token=abc").unwrap();
        let redacted = redact_error_url(&url, &auth);
        assert!(
            !redacted.contains("abc"),
            "common secret key redacted: {redacted}"
        );
        assert!(!redacted.contains("pw@"), "userinfo redacted: {redacted}");
    }

    #[test]
    fn test_substitute_context_substitutes_placeholders() {
        let mut ctx = HashMap::new();
        ctx.insert("org_id".to_string(), json!("acme"));
        ctx.insert("repo".to_string(), json!("myrepo"));
        let result =
            faucet_core::util::substitute_context("/orgs/{org_id}/repos/{repo}/issues", &ctx);
        assert_eq!(result, "/orgs/acme/repos/myrepo/issues");
    }

    #[test]
    fn test_substitute_context_no_placeholders() {
        let ctx = HashMap::new();
        let result = faucet_core::util::substitute_context("/api/users", &ctx);
        assert_eq!(result, "/api/users");
    }

    #[test]
    fn test_substitute_context_numeric_value() {
        let mut ctx = HashMap::new();
        ctx.insert("id".to_string(), json!(42));
        let result = faucet_core::util::substitute_context("/items/{id}", &ctx);
        assert_eq!(result, "/items/42");
    }

    #[test]
    fn test_parse_retry_after_valid() {
        let mut headers = HeaderMap::new();
        headers.insert(
            reqwest::header::RETRY_AFTER,
            reqwest::header::HeaderValue::from_static("30"),
        );
        assert_eq!(parse_retry_after(&headers), Duration::from_secs(30));
    }

    #[test]
    fn test_parse_retry_after_missing_defaults_to_60() {
        assert_eq!(
            parse_retry_after(&HeaderMap::new()),
            Duration::from_secs(60)
        );
    }

    #[test]
    fn test_parse_retry_after_non_numeric_defaults_to_60() {
        let mut headers = HeaderMap::new();
        headers.insert(
            reqwest::header::RETRY_AFTER,
            reqwest::header::HeaderValue::from_static("not-a-number"),
        );
        assert_eq!(parse_retry_after(&headers), Duration::from_secs(60));
    }

    #[test]
    fn test_parse_retry_after_http_date() {
        // RFC 7231 permits an HTTP-date form instead of delta-seconds.
        let future = std::time::SystemTime::now() + Duration::from_secs(7200);
        let date = httpdate::fmt_http_date(future);
        let mut headers = HeaderMap::new();
        headers.insert(
            reqwest::header::RETRY_AFTER,
            reqwest::header::HeaderValue::from_str(&date).unwrap(),
        );
        let d = parse_retry_after(&headers);
        // ~2 hours out — must not collapse to the 60s fallback.
        assert!(
            d > Duration::from_secs(3600),
            "expected ~2h from HTTP-date, got {d:?}"
        );
        assert!(
            d <= Duration::from_secs(7200),
            "should not exceed the target instant, got {d:?}"
        );
    }

    #[test]
    fn test_parse_retry_after_past_http_date_is_zero() {
        // A date already in the past → retry now (zero wait), not the fallback.
        let past = std::time::SystemTime::now() - Duration::from_secs(3600);
        let date = httpdate::fmt_http_date(past);
        let mut headers = HeaderMap::new();
        headers.insert(
            reqwest::header::RETRY_AFTER,
            reqwest::header::HeaderValue::from_str(&date).unwrap(),
        );
        assert_eq!(parse_retry_after(&headers), Duration::ZERO);
    }

    #[test]
    fn test_new_rejects_invalid_expiry_ratio_zero() {
        let config = RestStreamConfig::new("https://example.com", "/data").auth(Auth::OAuth2 {
            token_url: "https://auth.example.com/token".into(),
            client_id: "id".into(),
            client_secret: "secret".into(),
            scopes: vec![],
            expiry_ratio: 0.0,
        });
        let result = RestStream::new(config);
        assert!(result.is_err());
        assert!(matches!(result, Err(FaucetError::Auth(_))));
    }

    #[test]
    fn test_new_rejects_invalid_expiry_ratio_negative() {
        let config = RestStreamConfig::new("https://example.com", "/data").auth(Auth::OAuth2 {
            token_url: "https://auth.example.com/token".into(),
            client_id: "id".into(),
            client_secret: "secret".into(),
            scopes: vec![],
            expiry_ratio: -0.5,
        });
        assert!(RestStream::new(config).is_err());
    }

    #[test]
    fn test_new_rejects_invalid_expiry_ratio_above_one() {
        let config = RestStreamConfig::new("https://example.com", "/data").auth(Auth::OAuth2 {
            token_url: "https://auth.example.com/token".into(),
            client_id: "id".into(),
            client_secret: "secret".into(),
            scopes: vec![],
            expiry_ratio: 1.5,
        });
        assert!(RestStream::new(config).is_err());
    }

    #[test]
    fn test_new_accepts_valid_expiry_ratio() {
        let config = RestStreamConfig::new("https://example.com", "/data").auth(Auth::OAuth2 {
            token_url: "https://auth.example.com/token".into(),
            client_id: "id".into(),
            client_secret: "secret".into(),
            scopes: vec![],
            expiry_ratio: 1.0,
        });
        assert!(RestStream::new(config).is_ok());
    }

    #[test]
    fn test_new_with_no_auth_succeeds() {
        let config = RestStreamConfig::new("https://example.com", "/data");
        assert!(RestStream::new(config).is_ok());
    }

    #[test]
    fn test_new_with_timeout() {
        let config =
            RestStreamConfig::new("https://example.com", "/data").timeout(Duration::from_secs(10));
        assert!(RestStream::new(config).is_ok());
    }

    #[test]
    fn test_substitute_context_missing_placeholder_unchanged() {
        let mut ctx = HashMap::new();
        ctx.insert("org".to_string(), json!("acme"));
        let result = faucet_core::util::substitute_context("/items/{missing}", &ctx);
        assert_eq!(result, "/items/{missing}");
    }

    #[test]
    fn test_substitute_context_boolean_value() {
        let mut ctx = HashMap::new();
        ctx.insert("flag".to_string(), json!(true));
        let result = faucet_core::util::substitute_context("/items/{flag}", &ctx);
        assert_eq!(result, "/items/true");
    }

    #[test]
    fn rest_source_connector_name_is_rest() {
        use faucet_core::Source;
        let source = RestStream::new(RestStreamConfig::new("https://example.com", "/data"))
            .expect("minimal RestStream construction");
        assert_eq!(source.connector_name(), "rest");
    }

    #[test]
    fn dataset_uri_combines_base_and_path() {
        use faucet_core::Source;
        let source = RestStream::new(RestStreamConfig::new(
            "https://api.example.com",
            "/v1/users",
        ))
        .unwrap();
        assert_eq!(source.dataset_uri(), "https://api.example.com/v1/users");
    }

    #[test]
    fn dataset_uri_redacts_credentials() {
        use faucet_core::Source;
        let source = RestStream::new(RestStreamConfig::new(
            "https://user:secret@api.example.com",
            "/v1/data",
        ))
        .unwrap();
        assert_eq!(source.dataset_uri(), "https://api.example.com/v1/data");
    }
}

/// Mutual-TLS unit tests (#495). Lib-level so llvm-cov attributes coverage of
/// `apply_client_tls` / `build_identity` / the `new()` TLS branch reliably.
#[cfg(all(test, feature = "mtls"))]
mod mtls_tests {
    use super::*;
    use crate::config::TlsClientConfig;

    const CERT: &str = include_str!("../tests/fixtures/mtls/cert.pem");
    const KEY: &str = include_str!("../tests/fixtures/mtls/key.pem");

    fn pem() -> TlsClientConfig {
        TlsClientConfig {
            client_cert: Some(CERT.to_string()),
            client_key: Some(KEY.to_string()),
            ..Default::default()
        }
    }

    #[test]
    fn pem_identity_builds() {
        let cfg = RestStreamConfig::new("https://x.test", "/y").tls(pem());
        assert!(RestStream::new(cfg).is_ok());
    }

    #[test]
    fn min_version_branches_are_exercised() {
        // 1.2 is universally supported and must build.
        let mut tls = pem();
        tls.min_version = Some("1.2".into());
        assert!(RestStream::new(RestStreamConfig::new("https://x.test", "/y").tls(tls)).is_ok());
        // 1.3 exercises the other branch; some native-tls backends (e.g. macOS
        // SecureTransport) reject a 1.3 floor at client-build time, so only
        // require it not to panic.
        let mut tls = pem();
        tls.min_version = Some("1.3".into());
        let _ = RestStream::new(RestStreamConfig::new("https://x.test", "/y").tls(tls));
    }

    #[test]
    fn pkcs12_identity_builds() {
        let p12 = concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/tests/fixtures/mtls/identity.p12"
        );
        let tls = TlsClientConfig {
            client_identity_pkcs12: Some(p12.to_string()),
            pkcs12_password: Some("changeit".into()),
            ..Default::default()
        };
        let cfg = RestStreamConfig::new("https://x.test", "/y").tls(tls);
        assert!(RestStream::new(cfg).is_ok());
    }

    #[test]
    fn invalid_pem_errors_without_leaking_key() {
        let tls = TlsClientConfig {
            client_cert: Some("-----BEGIN CERTIFICATE-----\nbad\n-----END CERTIFICATE-----".into()),
            client_key: Some("SUPERSECRETKEY".into()),
            ..Default::default()
        };
        let cfg = RestStreamConfig::new("https://x.test", "/y").tls(tls);
        let err = RestStream::new(cfg)
            .map(|_| ())
            .expect_err("bad PEM must error");
        assert!(!err.to_string().contains("SUPERSECRETKEY"));
    }

    #[test]
    fn missing_pkcs12_file_errors() {
        let tls = TlsClientConfig {
            client_identity_pkcs12: Some("/no/such.p12".into()),
            pkcs12_password: Some("x".into()),
            ..Default::default()
        };
        let cfg = RestStreamConfig::new("https://x.test", "/y").tls(tls);
        assert!(RestStream::new(cfg).is_err());
    }
}