ff_sdk/

admin.rs

//! Admin REST client for operator-facing endpoints on `ff-server`.
//!
//! Wraps `POST /v1/admin/*` so downstream consumers (cairn-fabric)
//! don't hand-roll the HTTP call for admin surfaces like HMAC secret
//! rotation. Mirrors the server's wire types exactly — request
//! bodies and response shapes are defined against
//! [`ff_server::api`] + [`ff_server::server`] and kept 1:1 with the
//! producer.
//!
//! Authentication is Bearer token. Callers pick up the token from
//! wherever they hold it (`FF_API_TOKEN` env var is the common
//! pattern, but the SDK does not read env vars on the caller's
//! behalf — [`FlowFabricAdminClient::with_token`] accepts a
//! string-like token value (`&str` or `String`) via
//! `impl AsRef<str>`).

use std::sync::Arc;
use std::time::Duration;

use ff_core::contracts::RotateWaitpointHmacSecretOutcome;
use ff_core::engine_backend::EngineBackend;
// v0.12 PR-6: these imports only power the Valkey-typed
// `rotate_waitpoint_hmac_secret_all_partitions` helper at the bottom
// of the module; gated so the ungated module builds clean under
// `--no-default-features --features sqlite`.
#[cfg(feature = "valkey-default")]
use ff_core::contracts::RotateWaitpointHmacSecretArgs;
#[cfg(feature = "valkey-default")]
use ff_core::keys::IndexKeys;
#[cfg(feature = "valkey-default")]
use ff_core::partition::{Partition, PartitionFamily};
use serde::{Deserialize, Serialize};

use crate::SdkError;

/// Default per-request timeout. The server's own
/// `ROTATE_HTTP_TIMEOUT` is 120s; pick 130s client-side so the
/// client deadline is LATER than the server deadline and
/// operators see the structured 504 GATEWAY_TIMEOUT body rather
/// than a client-side timeout error.
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(130);

/// Grace window in ms that the embedded `rotate_waitpoint_secret`
/// path forwards to the backend primitive. Matches `ff-server`'s
/// default `FF_WAITPOINT_HMAC_GRACE_MS` (24 h) so tokens signed by
/// the outgoing kid remain valid for a full day after rotation
/// without the embedded-path hard-killing in-flight flows. HTTP
/// callers get whatever the server was configured with; embedded
/// callers get this pinned default.
pub const EMBEDDED_WAITPOINT_HMAC_GRACE_MS: u64 = 86_400_000;

/// Maximum grant TTL (ms) the embedded admin path accepts on
/// `claim_for_worker` / `issue_reclaim_grant`. Matches
/// `ff-server`'s `CLAIM_GRANT_TTL_MS_MAX` / `RECLAIM_GRANT_TTL_MS_MAX`
/// so the embedded transport rejects the same range the HTTP
/// transport does.
const EMBEDDED_GRANT_TTL_MS_MAX: u64 = 60_000;

/// Client for FlowFabric admin primitives — backend-agnostic facade
/// (v0.13 SC-10 ergonomics).
///
/// Two construction shapes:
///
/// * [`FlowFabricAdminClient::new`] / [`FlowFabricAdminClient::with_token`]
///   — HTTP transport targeting a running `ff-server`.
/// * [`FlowFabricAdminClient::connect_with`] — embedded transport
///   that dispatches directly through an `Arc<dyn EngineBackend>`.
///   No `ff-server` required; works under `FF_DEV_MODE=1` + SQLite
///   and in any in-process deployment.
///
/// The public method surface is identical across both transports;
/// consumers choose at construction time. Admin methods that have
/// no backend-trait equivalent return
/// [`SdkError::AdminApi`] with status 503 on the embedded path —
/// today every method on this client maps cleanly, so this fallback
/// is only reached if a future admin primitive lands HTTP-first.
#[derive(Debug, Clone)]
pub struct FlowFabricAdminClient {
    transport: AdminTransport,
}

/// Internal discriminator between the HTTP and embedded transports.
/// Private by design — the public API is uniform across both shapes
/// (see the [`FlowFabricAdminClient`] type-level docs).
#[derive(Clone)]
enum AdminTransport {
    Http {
        http: reqwest::Client,
        base_url: String,
    },
    Embedded(Arc<dyn EngineBackend>),
}

impl std::fmt::Debug for AdminTransport {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            AdminTransport::Http { base_url, .. } => f
                .debug_struct("Http")
                .field("base_url", base_url)
                .finish_non_exhaustive(),
            AdminTransport::Embedded(backend) => f
                .debug_struct("Embedded")
                .field("backend", &backend.backend_label())
                .finish(),
        }
    }
}

impl FlowFabricAdminClient {
    /// Build a client without auth. Suitable for a dev ff-server
    /// whose `api_token` is unconfigured. Production deployments
    /// should use [`with_token`](Self::with_token).
    pub fn new(base_url: impl Into<String>) -> Result<Self, SdkError> {
        let http = reqwest::Client::builder()
            .timeout(DEFAULT_TIMEOUT)
            .build()
            .map_err(|e| SdkError::Http {
                source: e,
                context: "build reqwest::Client".into(),
            })?;
        Ok(Self {
            transport: AdminTransport::Http {
                http,
                base_url: normalize_base_url(base_url.into()),
            },
        })
    }

    /// Build a client that dispatches admin primitives directly
    /// through an `Arc<dyn EngineBackend>`, bypassing HTTP entirely.
    ///
    /// # When to use
    ///
    /// * `FF_DEV_MODE=1` SQLite scenarios where no `ff-server` is
    ///   running.
    /// * In-process / embedded deployments that hold a backend
    ///   handle already (e.g. tests, examples, scheduler-hosting
    ///   binaries).
    ///
    /// # Semantic parity
    ///
    /// Each method on [`FlowFabricAdminClient`] dispatches to the
    /// equivalent `EngineBackend` trait method (see the RFC-024 /
    /// RFC-017 admin surfaces). Validation **rules** + request-body
    /// translation mirror the server-side handler in `ff-server` so
    /// callers get the same accept / reject behaviour across
    /// transports. Note: the exact [`SdkError`] variant differs —
    /// embedded-path validation rejects surface as [`SdkError::Config`]
    /// (no HTTP round-trip) while HTTP returns [`SdkError::AdminApi`]
    /// with status `400`. Callers that need to distinguish a 4xx from
    /// a transport fault should use [`SdkError::is_retryable`] or
    /// match on `Config` + `AdminApi` together rather than relying on
    /// a single variant.
    ///
    /// `EngineError::Unavailable` from the backend — emitted by
    /// backends that have not implemented a given method — is mapped
    /// to [`SdkError::AdminApi`] with `status = 503` and
    /// `kind = Some("unavailable")` so callers see a uniform
    /// admin-error surface across transports.
    ///
    /// # Divergence from the HTTP transport
    ///
    /// * `rotate_waitpoint_secret` forwards
    ///   [`EMBEDDED_WAITPOINT_HMAC_GRACE_MS`] (24 h, matching
    ///   `ff-server`'s default `FF_WAITPOINT_HMAC_GRACE_MS`) as the
    ///   per-partition grace window. The HTTP transport reads the
    ///   server's env-configured value; the embedded client has no
    ///   config surface so it pins the documented default.
    /// * No single-writer admin semaphore, no audit-log emission.
    ///   These are `ff-server` responsibilities; embedded consumers
    ///   wanting them bring their own gate.
    pub fn connect_with(backend: Arc<dyn EngineBackend>) -> Self {
        Self {
            transport: AdminTransport::Embedded(backend),
        }
    }

    /// Build a client that sends `Authorization: Bearer <token>` on
    /// every request. The token is passed by value so the caller
    /// retains ownership policy (e.g. zeroize on drop at the
    /// caller side); the SDK only reads it.
    ///
    /// # Empty-token guard
    ///
    /// An empty or all-whitespace `token` returns
    /// [`SdkError::Config`] instead of silently constructing
    /// `Authorization: Bearer ` (which the server rejects with
    /// 401, leaving the operator chasing a "why is auth broken"
    /// ghost). Common source: `FF_ADMIN_TOKEN=""` in a shell
    /// where the var was meant to be set; the unset-expansion is
    /// the empty string. Prefer an obvious error at construction
    /// over a silent 401 at first request.
    ///
    /// If the caller genuinely wants an unauthenticated client
    /// (dev ff-server without `api_token` configured), use
    /// [`FlowFabricAdminClient::new`] instead.
    pub fn with_token(
        base_url: impl Into<String>,
        token: impl AsRef<str>,
    ) -> Result<Self, SdkError> {
        let token_str = token.as_ref();
        if token_str.trim().is_empty() {
            return Err(SdkError::Config {
                context: "admin_client".into(),
                field: Some("bearer_token".into()),
                message: "is empty or all-whitespace; use \
                          FlowFabricAdminClient::new for unauthenticated access"
                    .into(),
            });
        }
        let mut headers = reqwest::header::HeaderMap::new();
        let mut auth_value =
            reqwest::header::HeaderValue::from_str(&format!("Bearer {}", token_str)).map_err(
                |_| SdkError::Config {
                    context: "admin_client".into(),
                    field: Some("bearer_token".into()),
                    message: "contains characters not valid in an HTTP header".into(),
                },
            )?;
        // Mark Authorization as sensitive so it doesn't appear in
        // reqwest's Debug output / logs.
        auth_value.set_sensitive(true);
        headers.insert(reqwest::header::AUTHORIZATION, auth_value);

        let http = reqwest::Client::builder()
            .timeout(DEFAULT_TIMEOUT)
            .default_headers(headers)
            .build()
            .map_err(|e| SdkError::Http {
                source: e,
                context: "build reqwest::Client".into(),
            })?;
        Ok(Self {
            transport: AdminTransport::Http {
                http,
                base_url: normalize_base_url(base_url.into()),
            },
        })
    }

    /// POST `/v1/workers/{worker_id}/claim` — scheduler-routed claim.
    ///
    /// Batch C item 2 PR-B. Swaps the SDK's direct-Valkey claim for a
    /// server-side one: the request carries lane + identity +
    /// capabilities + grant TTL; the server runs budget, quota, and
    /// capability admission via `ff_scheduler::Scheduler::claim_for_worker`
    /// and returns a `ClaimGrant` on success.
    ///
    /// Returns `Ok(None)` when the server responds 204 No Content
    /// (no eligible execution on the lane). Callers that want to keep
    /// polling should back off per their claim cadence.
    pub async fn claim_for_worker(
        &self,
        req: ClaimForWorkerRequest,
    ) -> Result<Option<ClaimForWorkerResponse>, SdkError> {
        match &self.transport {
            AdminTransport::Http { http, base_url } => {
                claim_for_worker_http(http, base_url, req).await
            }
            AdminTransport::Embedded(backend) => {
                claim_for_worker_embedded(backend.as_ref(), req).await
            }
        }
    }

    /// Rotate the waitpoint HMAC secret on the server.
    ///
    /// Promotes the currently-installed kid to `previous_kid`
    /// (accepted for the server's configured
    /// `FF_WAITPOINT_HMAC_GRACE_MS` window) and installs
    /// `new_secret_hex` under `new_kid` as the new current. Fans
    /// out across every execution partition. Idempotent: re-running
    /// with the same `(new_kid, new_secret_hex)` converges.
    ///
    /// The server returns 200 if at least one partition rotated OR
    /// at least one partition was already rotating under a
    /// concurrent request. See `RotateWaitpointSecretResponse`
    /// fields for the breakdown.
    ///
    /// # Errors
    ///
    /// * [`SdkError::AdminApi`] — non-2xx response (400 invalid
    ///   input, 401 missing/bad bearer, 429 concurrent rotate,
    ///   500 all partitions failed, 504 server-side timeout).
    /// * [`SdkError::Http`] — transport error (connect, body
    ///   decode, client-side timeout).
    ///
    /// # Retry semantics
    ///
    /// Rotation is idempotent on the same `(new_kid,
    /// new_secret_hex)` so retries are SAFE even on 504s or
    /// partial failures.
    pub async fn rotate_waitpoint_secret(
        &self,
        req: RotateWaitpointSecretRequest,
    ) -> Result<RotateWaitpointSecretResponse, SdkError> {
        match &self.transport {
            AdminTransport::Http { http, base_url } => {
                rotate_waitpoint_secret_http(http, base_url, req).await
            }
            AdminTransport::Embedded(backend) => {
                rotate_waitpoint_secret_embedded(backend.as_ref(), req).await
            }
        }
    }

    /// Read the raw HMAC `waitpoint_token` for a specific
    /// `(execution_id, waitpoint_id)` pair.
    ///
    /// # Operator-only
    ///
    /// The sibling `list_pending_waitpoints` API intentionally sanitises
    /// this field (RFC-017 Stage E4 / v0.8.0 §8) — consumers correlate
    /// via `(token_kid, token_fingerprint)` and normally obtain the raw
    /// token from their own worker's `SuspendOutcome` at suspend time.
    /// This admin method re-exposes the token behind the server's
    /// bearer-auth layer so operator tooling (approval CLIs,
    /// external-callback bridges) can fetch a delivery credential
    /// programmatically instead of copy-pasting from worker logs.
    ///
    /// Deployments MUST run `ff-server` with `api_token` configured
    /// when exposing this endpoint to untrusted networks. The embedded
    /// transport has no auth boundary — access is gated by whoever
    /// holds the `Arc<dyn EngineBackend>`.
    ///
    /// # Returns
    ///
    /// * `Ok(Some(token))` — HMAC token string suitable for
    ///   `DeliverSignalArgs::waitpoint_token` / `/signal` request body.
    /// * `Ok(None)` — waitpoint is unknown, consumed, expired, or the
    ///   stored token column is empty.
    /// * `Err(SdkError::AdminApi { status: 503, kind: Some("unavailable") })`
    ///   — the backend (e.g. an out-of-tree implementation) has not
    ///   overridden `EngineBackend::read_waitpoint_token`.
    pub async fn read_waitpoint_token(
        &self,
        execution_id: &ff_core::types::ExecutionId,
        waitpoint_id: &ff_core::types::WaitpointId,
    ) -> Result<Option<String>, SdkError> {
        match &self.transport {
            AdminTransport::Http { http, base_url } => {
                read_waitpoint_token_http(http, base_url, execution_id, waitpoint_id).await
            }
            AdminTransport::Embedded(backend) => {
                read_waitpoint_token_embedded(backend.as_ref(), execution_id, waitpoint_id).await
            }
        }
    }

    /// Request a lease-reclaim grant for an execution in
    /// `lease_expired_reclaimable` or `lease_revoked` state (RFC-024
    /// §3.5).
    ///
    /// Routes `POST /v1/executions/{execution_id}/reclaim`. The
    /// ff-server handler dispatches through the `EngineBackend` trait
    /// to whichever backend the server was started with (Valkey /
    /// Postgres / SQLite).
    ///
    /// # worker_capabilities (RFC-024 §3.2 B-2)
    ///
    /// The request body carries `worker_capabilities`. Consumers typically
    /// source these from their registered worker's configured
    /// `WorkerConfig::capabilities`. Admission compares
    /// `worker_capabilities` against the execution's
    /// `required_capabilities` (persisted on `exec_core` at
    /// `create_execution` time from
    /// `ExecutionPolicy.routing_requirements.required_capabilities`);
    /// any required capability missing from the worker set surfaces as
    /// `IssueReclaimGrantResponse::NotReclaimable { detail:
    /// "capability_mismatch: <missing csv>" }` (Lua
    /// `ff_issue_reclaim_grant`, `crates/ff-script/src/flowfabric.lua`
    /// §3969-3982; sqlite/PG backends mirror the check). The SDK does
    /// not re-read worker state automatically — admin clients are not
    /// bound to a worker — so the consumer threads the capabilities
    /// through at call-time.
    ///
    /// `capability_hash` is NOT consulted for admission; it is stored
    /// verbatim on the grant hash for audit / downstream observability
    /// only.
    ///
    /// # Consumer flow (RFC-024 §4.4)
    ///
    /// 1. Consumer's `POST /v1/runs/:id/complete` returns
    ///    `lease_expired`.
    /// 2. Consumer calls this method; handles
    ///    [`IssueReclaimGrantResponse::Granted`] → builds a
    ///    [`ff_core::contracts::ReclaimGrant`] via
    ///    [`IssueReclaimGrantResponse::into_grant`].
    /// 3. Consumer passes the grant to
    ///    [`crate::FlowFabricWorker::claim_from_reclaim_grant`] along
    ///    with a fresh [`ff_core::contracts::ReclaimExecutionArgs`];
    ///    the new attempt is minted with `HandleKind::Reclaimed`.
    /// 4. Consumer drives terminal writes on the fresh lease.
    ///
    /// # Errors
    ///
    /// * [`SdkError::AdminApi`] — non-2xx response. 404 when the
    ///   execution does not exist; 400 on malformed `execution_id` or
    ///   body.
    /// * [`SdkError::Http`] — transport error (connect, body
    ///   decode, client-side timeout).
    ///
    /// # Retry semantics
    ///
    /// Idempotent on the Lua side: repeated calls against an execution
    /// already re-leased (a concurrent reclaim beat this one) surface
    /// as `NotReclaimable`. Safe to retry on transport faults.
    pub async fn issue_reclaim_grant(
        &self,
        execution_id: &str,
        req: IssueReclaimGrantRequest,
    ) -> Result<IssueReclaimGrantResponse, SdkError> {
        match &self.transport {
            AdminTransport::Http { http, base_url } => {
                issue_reclaim_grant_http(http, base_url, execution_id, req).await
            }
            AdminTransport::Embedded(backend) => {
                issue_reclaim_grant_embedded(backend.as_ref(), execution_id, req).await
            }
        }
    }
}

// ── HTTP-transport helpers (private) ─────────────────────────────────

async fn claim_for_worker_http(
    http: &reqwest::Client,
    base_url: &str,
    req: ClaimForWorkerRequest,
) -> Result<Option<ClaimForWorkerResponse>, SdkError> {
    // Percent-encode `worker_id` in the URL path — `WorkerId` is a
    // free-form string (could contain `/`, spaces, `%`, etc.) and
    // splicing it verbatim would produce malformed URLs or
    // misrouted paths. `Url::path_segments_mut().push` handles the
    // encoding natively.
    let mut url = reqwest::Url::parse(base_url).map_err(|e| SdkError::Config {
        context: "admin_client: claim_for_worker".into(),
        field: Some("base_url".into()),
        message: format!("invalid base_url '{}': {e}", base_url),
    })?;
    {
        let mut segs = url.path_segments_mut().map_err(|_| SdkError::Config {
            context: "admin_client: claim_for_worker".into(),
            field: Some("base_url".into()),
            message: format!("base_url cannot be a base URL: '{}'", base_url),
        })?;
        segs.extend(&["v1", "workers", &req.worker_id, "claim"]);
    }
    let url = url.to_string();
    let resp = http
        .post(&url)
        .json(&req)
        .send()
        .await
        .map_err(|e| SdkError::Http {
            source: e,
            context: "POST /v1/workers/{worker_id}/claim".into(),
        })?;

    let status = resp.status();
    if status == reqwest::StatusCode::NO_CONTENT {
        return Ok(None);
    }
    if status.is_success() {
        return resp
            .json::<ClaimForWorkerResponse>()
            .await
            .map(Some)
            .map_err(|e| SdkError::Http {
                source: e,
                context: "decode claim_for_worker response body".into(),
            });
    }

    // Error path — mirror rotate_waitpoint_secret's ErrorBody decode.
    let status_u16 = status.as_u16();
    let raw = resp.text().await.map_err(|e| SdkError::Http {
        source: e,
        context: format!("read claim_for_worker error body (status {status_u16})"),
    })?;
    let parsed = serde_json::from_str::<AdminErrorBody>(&raw).ok();
    Err(SdkError::AdminApi {
        status: status_u16,
        message: parsed
            .as_ref()
            .map(|b| b.error.clone())
            .unwrap_or_else(|| raw.clone()),
        kind: parsed.as_ref().and_then(|b| b.kind.clone()),
        retryable: parsed.as_ref().and_then(|b| b.retryable),
        raw_body: raw,
    })
}

async fn rotate_waitpoint_secret_http(
    http: &reqwest::Client,
    base_url: &str,
    req: RotateWaitpointSecretRequest,
) -> Result<RotateWaitpointSecretResponse, SdkError> {
    let url = format!("{}/v1/admin/rotate-waitpoint-secret", base_url);
    let resp = http
        .post(&url)
        .json(&req)
        .send()
        .await
        .map_err(|e| SdkError::Http {
            source: e,
            context: "POST /v1/admin/rotate-waitpoint-secret".into(),
        })?;

    let status = resp.status();
    if status.is_success() {
        return resp
            .json::<RotateWaitpointSecretResponse>()
            .await
            .map_err(|e| SdkError::Http {
                source: e,
                context: "decode rotate-waitpoint-secret response body".into(),
            });
    }

    // Non-2xx: parse the server's ErrorBody if we can, fall
    // back to a raw body otherwise. Propagate body-read
    // transport errors as Http rather than silently flattening
    // them into `AdminApi { raw_body: "" }` — a connection drop
    // mid-body-read is a transport fault, not an API-layer
    // reject, and misclassifying it strips `is_retryable`'s
    // timeout/connect signal from the caller.
    let status_u16 = status.as_u16();
    let raw = resp.text().await.map_err(|e| SdkError::Http {
        source: e,
        context: format!(
            "read rotate-waitpoint-secret error response body (status {status_u16})"
        ),
    })?;
    let parsed = serde_json::from_str::<AdminErrorBody>(&raw).ok();
    Err(SdkError::AdminApi {
        status: status_u16,
        message: parsed
            .as_ref()
            .map(|b| b.error.clone())
            .unwrap_or_else(|| raw.clone()),
        kind: parsed.as_ref().and_then(|b| b.kind.clone()),
        retryable: parsed.as_ref().and_then(|b| b.retryable),
        raw_body: raw,
    })
}

async fn issue_reclaim_grant_http(
    http: &reqwest::Client,
    base_url: &str,
    execution_id: &str,
    req: IssueReclaimGrantRequest,
) -> Result<IssueReclaimGrantResponse, SdkError> {
    // Percent-encode `execution_id` in the URL path — the id is a
    // free-form string and splicing verbatim would produce
    // malformed URLs. Mirrors `claim_for_worker`'s handling.
    let mut url = reqwest::Url::parse(base_url).map_err(|e| SdkError::Config {
        context: "admin_client: issue_reclaim_grant".into(),
        field: Some("base_url".into()),
        message: format!("invalid base_url '{}': {e}", base_url),
    })?;
    {
        let mut segs = url.path_segments_mut().map_err(|_| SdkError::Config {
            context: "admin_client: issue_reclaim_grant".into(),
            field: Some("base_url".into()),
            message: format!("base_url cannot be a base URL: '{}'", base_url),
        })?;
        segs.extend(&["v1", "executions", execution_id, "reclaim"]);
    }
    let url = url.to_string();
    let resp = http
        .post(&url)
        .json(&req)
        .send()
        .await
        .map_err(|e| SdkError::Http {
            source: e,
            context: "POST /v1/executions/{id}/reclaim".into(),
        })?;

    let status = resp.status();
    if status.is_success() {
        return resp
            .json::<IssueReclaimGrantResponse>()
            .await
            .map_err(|e| SdkError::Http {
                source: e,
                context: "decode issue_reclaim_grant response body".into(),
            });
    }

    let status_u16 = status.as_u16();
    let raw = resp.text().await.map_err(|e| SdkError::Http {
        source: e,
        context: format!(
            "read issue_reclaim_grant error body (status {status_u16})"
        ),
    })?;
    let parsed = serde_json::from_str::<AdminErrorBody>(&raw).ok();
    Err(SdkError::AdminApi {
        status: status_u16,
        message: parsed
            .as_ref()
            .map(|b| b.error.clone())
            .unwrap_or_else(|| raw.clone()),
        kind: parsed.as_ref().and_then(|b| b.kind.clone()),
        retryable: parsed.as_ref().and_then(|b| b.retryable),
        raw_body: raw,
    })
}

// ── Embedded-transport helpers (private) ─────────────────────────────
//
// Dispatch directly through the `EngineBackend` trait. The request
// body validation mirrors the server-side handler in
// `ff-server::api`. Translation between the wire DTOs and the core
// `contracts::*` types lives here so consumers get identical
// surfaces across transports.

/// Validate a free-form identifier the same way `ff-server`'s
/// `validate_identifier` does: non-empty, ≤256 bytes, no whitespace
/// or control chars. Embedded-transport callers hit this before the
/// request reaches the backend so invalid identifiers cannot leak
/// past the SDK's parity guarantee.
fn validate_admin_identifier(
    op: &'static str,
    field: &'static str,
    value: &str,
) -> Result<(), SdkError> {
    if value.is_empty() {
        return Err(SdkError::Config {
            context: format!("admin_client: {op}"),
            field: Some(field.into()),
            message: "must not be empty".into(),
        });
    }
    if value.len() > 256 {
        return Err(SdkError::Config {
            context: format!("admin_client: {op}"),
            field: Some(field.into()),
            message: format!("exceeds 256 bytes (got {})", value.len()),
        });
    }
    if value.chars().any(|c| c.is_control() || c.is_whitespace()) {
        return Err(SdkError::Config {
            context: format!("admin_client: {op}"),
            field: Some(field.into()),
            message: "must not contain whitespace or control characters".into(),
        });
    }
    Ok(())
}

/// Bounded grant-TTL check mirroring `ff-server`'s
/// `1..=CLAIM_GRANT_TTL_MS_MAX`. Shared between `claim_for_worker`
/// and `issue_reclaim_grant` embedded paths.
fn validate_admin_grant_ttl(op: &'static str, grant_ttl_ms: u64) -> Result<(), SdkError> {
    if grant_ttl_ms == 0 || grant_ttl_ms > EMBEDDED_GRANT_TTL_MS_MAX {
        return Err(SdkError::Config {
            context: format!("admin_client: {op}"),
            field: Some("grant_ttl_ms".into()),
            message: format!("must be in 1..={EMBEDDED_GRANT_TTL_MS_MAX}"),
        });
    }
    Ok(())
}

/// Map an `EngineError` from a backend call into the `SdkError::AdminApi`
/// surface so embedded and HTTP transports emit the same shape. `Unavailable`
/// becomes 503 with kind `"unavailable"`; every other engine error bubbles
/// up via `SdkError::Engine`.
fn engine_err_to_admin(err: ff_core::engine_error::EngineError, op: &str) -> SdkError {
    if let ff_core::engine_error::EngineError::Unavailable { op: backend_op } = &err {
        return SdkError::AdminApi {
            status: 503,
            message: format!(
                "{op}: backend does not implement '{backend_op}' on this transport"
            ),
            kind: Some("unavailable".to_owned()),
            retryable: Some(false),
            raw_body: String::new(),
        };
    }
    SdkError::Engine(Box::new(err))
}

async fn claim_for_worker_embedded(
    backend: &dyn EngineBackend,
    req: ClaimForWorkerRequest,
) -> Result<Option<ClaimForWorkerResponse>, SdkError> {
    // Mirror ff-server's validation + translation. Errors surface as
    // SdkError::Config so consumers see validation faults loud rather
    // than as backend transport errors.
    let lane_id = ff_core::types::LaneId::try_new(req.lane_id).map_err(|e| SdkError::Config {
        context: "admin_client: claim_for_worker".into(),
        field: Some("lane_id".into()),
        message: e.to_string(),
    })?;
    validate_admin_identifier("claim_for_worker", "worker_id", &req.worker_id)?;
    validate_admin_identifier(
        "claim_for_worker",
        "worker_instance_id",
        &req.worker_instance_id,
    )?;
    validate_admin_grant_ttl("claim_for_worker", req.grant_ttl_ms)?;
    let caps: std::collections::BTreeSet<String> = req.capabilities.into_iter().collect();
    let args = ff_core::contracts::ClaimForWorkerArgs::new(
        lane_id,
        ff_core::types::WorkerId::new(req.worker_id),
        ff_core::types::WorkerInstanceId::new(req.worker_instance_id),
        caps,
        req.grant_ttl_ms,
    );
    match backend
        .claim_for_worker(args)
        .await
        .map_err(|e| engine_err_to_admin(e, "claim_for_worker"))?
    {
        ff_core::contracts::ClaimForWorkerOutcome::NoWork => Ok(None),
        ff_core::contracts::ClaimForWorkerOutcome::Granted(grant) => {
            Ok(Some(ClaimForWorkerResponse {
                execution_id: grant.execution_id.to_string(),
                partition_key: grant.partition_key,
                grant_key: grant.grant_key,
                expires_at_ms: grant.expires_at_ms,
            }))
        }
        // `ClaimForWorkerOutcome` is `#[non_exhaustive]`; mirror
        // ff-server's 503 policy on unknown variants.
        _ => Err(SdkError::AdminApi {
            status: 503,
            message: "claim_for_worker: backend returned a non-exhaustive outcome this SDK build does not understand".to_owned(),
            kind: Some("unknown_outcome".to_owned()),
            retryable: Some(false),
            raw_body: String::new(),
        }),
    }
}

async fn issue_reclaim_grant_embedded(
    backend: &dyn EngineBackend,
    execution_id: &str,
    req: IssueReclaimGrantRequest,
) -> Result<IssueReclaimGrantResponse, SdkError> {
    let exec_id = ff_core::types::ExecutionId::parse(execution_id).map_err(|e| SdkError::Config {
        context: "admin_client: issue_reclaim_grant".into(),
        field: Some("execution_id".into()),
        message: e.to_string(),
    })?;
    let lane_id = ff_core::types::LaneId::try_new(req.lane_id).map_err(|e| SdkError::Config {
        context: "admin_client: issue_reclaim_grant".into(),
        field: Some("lane_id".into()),
        message: e.to_string(),
    })?;
    validate_admin_identifier("issue_reclaim_grant", "worker_id", &req.worker_id)?;
    validate_admin_identifier(
        "issue_reclaim_grant",
        "worker_instance_id",
        &req.worker_instance_id,
    )?;
    validate_admin_grant_ttl("issue_reclaim_grant", req.grant_ttl_ms)?;
    let caps: std::collections::BTreeSet<String> = req.worker_capabilities.into_iter().collect();
    let args = ff_core::contracts::IssueReclaimGrantArgs::new(
        exec_id,
        ff_core::types::WorkerId::new(req.worker_id),
        ff_core::types::WorkerInstanceId::new(req.worker_instance_id),
        lane_id,
        req.capability_hash,
        req.grant_ttl_ms,
        req.route_snapshot_json,
        req.admission_summary,
        caps,
        ff_core::types::TimestampMs::now(),
    );
    match backend
        .issue_reclaim_grant(args)
        .await
        .map_err(|e| engine_err_to_admin(e, "issue_reclaim_grant"))?
    {
        ff_core::contracts::IssueReclaimGrantOutcome::Granted(grant) => {
            Ok(IssueReclaimGrantResponse::Granted {
                execution_id: grant.execution_id.to_string(),
                partition_key: grant.partition_key,
                grant_key: grant.grant_key,
                expires_at_ms: grant.expires_at_ms,
                lane_id: grant.lane_id.as_str().to_owned(),
            })
        }
        ff_core::contracts::IssueReclaimGrantOutcome::NotReclaimable { execution_id, detail } => {
            Ok(IssueReclaimGrantResponse::NotReclaimable {
                execution_id: execution_id.to_string(),
                detail,
            })
        }
        ff_core::contracts::IssueReclaimGrantOutcome::ReclaimCapExceeded {
            execution_id,
            reclaim_count,
        } => Ok(IssueReclaimGrantResponse::ReclaimCapExceeded {
            execution_id: execution_id.to_string(),
            reclaim_count,
        }),
        _ => Err(SdkError::AdminApi {
            status: 503,
            message: "issue_reclaim_grant: backend returned a non-exhaustive outcome this SDK build does not understand".to_owned(),
            kind: Some("unknown_outcome".to_owned()),
            retryable: Some(false),
            raw_body: String::new(),
        }),
    }
}

async fn read_waitpoint_token_http(
    http: &reqwest::Client,
    base_url: &str,
    execution_id: &ff_core::types::ExecutionId,
    waitpoint_id: &ff_core::types::WaitpointId,
) -> Result<Option<String>, SdkError> {
    // Percent-encode both path segments: execution_id carries `{fp:N}:`
    // hash-tag punctuation; waitpoint_id is a UUID but be defensive.
    let mut url = reqwest::Url::parse(base_url).map_err(|e| SdkError::Config {
        context: "admin_client: read_waitpoint_token".into(),
        field: Some("base_url".into()),
        message: format!("invalid base_url '{}': {e}", base_url),
    })?;
    url.path_segments_mut()
        .map_err(|_| SdkError::Config {
            context: "admin_client: read_waitpoint_token".into(),
            field: Some("base_url".into()),
            message: format!("base_url '{}' cannot be a base", base_url),
        })?
        .extend(&[
            "v1",
            "executions",
            execution_id.as_str(),
            "waitpoints",
            &waitpoint_id.to_string(),
            "token",
        ]);

    let resp = http
        .get(url.clone())
        .send()
        .await
        .map_err(|e| SdkError::Http {
            source: e,
            context: format!("GET {url}"),
        })?;
    let status = resp.status();
    if status == reqwest::StatusCode::NOT_FOUND {
        return Ok(None);
    }
    if status.is_success() {
        #[derive(Deserialize)]
        struct Body {
            token: String,
        }
        let body: Body = resp.json().await.map_err(|e| SdkError::Http {
            source: e,
            context: "decode read_waitpoint_token response body".into(),
        })?;
        return Ok(Some(body.token));
    }

    let status_u16 = status.as_u16();
    let raw = resp.text().await.map_err(|e| SdkError::Http {
        source: e,
        context: format!("read read_waitpoint_token error body (status {status_u16})"),
    })?;
    let parsed = serde_json::from_str::<AdminErrorBody>(&raw).ok();
    Err(SdkError::AdminApi {
        status: status_u16,
        message: parsed
            .as_ref()
            .map(|b| b.error.clone())
            .unwrap_or_else(|| raw.clone()),
        kind: parsed.as_ref().and_then(|b| b.kind.clone()),
        retryable: parsed.as_ref().and_then(|b| b.retryable),
        raw_body: raw,
    })
}

async fn read_waitpoint_token_embedded(
    backend: &dyn EngineBackend,
    execution_id: &ff_core::types::ExecutionId,
    waitpoint_id: &ff_core::types::WaitpointId,
) -> Result<Option<String>, SdkError> {
    let partition =
        ff_core::partition::PartitionKey::from(&ff_core::partition::Partition {
            family: ff_core::partition::PartitionFamily::Flow,
            index: execution_id.partition(),
        });
    backend
        .read_waitpoint_token(partition, waitpoint_id)
        .await
        .map_err(|e| engine_err_to_admin(e, "read_waitpoint_token"))
}

async fn rotate_waitpoint_secret_embedded(
    backend: &dyn EngineBackend,
    req: RotateWaitpointSecretRequest,
) -> Result<RotateWaitpointSecretResponse, SdkError> {
    // Validation mirrors `ff-server::Server::rotate_waitpoint_secret`
    // so the embedded and HTTP transports reject the same invalid
    // inputs.
    if req.new_kid.is_empty() || req.new_kid.contains(':') {
        return Err(SdkError::Config {
            context: "admin_client: rotate_waitpoint_secret".into(),
            field: Some("new_kid".into()),
            message: "must be non-empty and must not contain ':'".into(),
        });
    }
    if req.new_secret_hex.is_empty()
        || !req.new_secret_hex.len().is_multiple_of(2)
        || !req.new_secret_hex.chars().all(|c| c.is_ascii_hexdigit())
    {
        return Err(SdkError::Config {
            context: "admin_client: rotate_waitpoint_secret".into(),
            field: Some("new_secret_hex".into()),
            message: "must be a non-empty even-length hex string".into(),
        });
    }
    // Embedded consumers have no config surface from which to read
    // the per-deployment grace window, so pin the documented default
    // matching `ff-server`'s `FF_WAITPOINT_HMAC_GRACE_MS` (24 h). See
    // `EMBEDDED_WAITPOINT_HMAC_GRACE_MS` + the `connect_with` rustdoc
    // for the rationale and divergence-from-HTTP notes. Unlike the
    // HTTP handler, this path does not enforce the 120 s end-to-end
    // timeout (no HTTP deadline to honour) and does not emit the
    // `audit`-target `waitpoint_hmac_rotation_*` events (those are
    // server-owned operator signals). Rotation is idempotent on the
    // same (new_kid, new_secret_hex) pair, so retries remain safe.
    let args = ff_core::contracts::RotateWaitpointHmacSecretAllArgs::new(
        req.new_kid.clone(),
        req.new_secret_hex,
        EMBEDDED_WAITPOINT_HMAC_GRACE_MS,
    );
    let result = backend
        .rotate_waitpoint_hmac_secret_all(args)
        .await
        .map_err(|e| engine_err_to_admin(e, "rotate_waitpoint_secret"))?;

    // Collapse the per-partition entries into the HTTP response shape
    // the server emits — rotated count + failed indices + echoed
    // new_kid — so consumers see identical return values across
    // transports.
    let mut rotated: u16 = 0;
    let mut failed: Vec<u16> = Vec::new();
    for entry in &result.entries {
        match &entry.result {
            Ok(_) => {
                rotated = rotated.saturating_add(1);
            }
            Err(_) => failed.push(entry.partition),
        }
    }
    Ok(RotateWaitpointSecretResponse {
        rotated,
        failed,
        new_kid: req.new_kid,
    })
}

/// Request body for `POST /v1/executions/{execution_id}/reclaim`
/// (RFC-024 §3.5).
///
/// Mirrors `ff_server::api::IssueReclaimGrantBody` 1:1. The
/// `execution_id` goes in the URL path, not the body.
#[derive(Debug, Clone, Serialize)]
pub struct IssueReclaimGrantRequest {
    /// Worker identity requesting the grant. The Lua
    /// `ff_reclaim_execution` validates grant consumption via
    /// `grant.worker_id == args.worker_id` (RFC-024 §4.4) — the
    /// worker consuming the grant must match this value.
    pub worker_id: String,
    /// Worker-instance identity. Informational at grant-issuance
    /// time; stored on the grant so consumers can correlate events.
    pub worker_instance_id: String,
    /// Lane the execution belongs to. Needed by
    /// `ff_issue_reclaim_grant` for `KEYS[*]` construction.
    pub lane_id: String,
    /// Opaque capability-hash token stored verbatim on the issued
    /// grant for audit / downstream observability. NOT used for
    /// admission — admission compares `worker_capabilities` against
    /// the execution's `required_capabilities` (see the
    /// [`FlowFabricAdminClient::issue_reclaim_grant`] rustdoc).
    /// `None` leaves the field empty on the grant.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub capability_hash: Option<String>,
    /// Grant TTL in milliseconds. Bounded server-side.
    pub grant_ttl_ms: u64,
    /// Route snapshot JSON carried onto the grant for audit.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub route_snapshot_json: Option<String>,
    /// Admission summary string carried onto the grant for audit.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub admission_summary: Option<String>,
    /// Worker capability tokens. Consumers typically source these
    /// from their registered worker's `WorkerConfig::capabilities`
    /// (see [`FlowFabricAdminClient::issue_reclaim_grant`] rustdoc
    /// for the override contract).
    #[serde(default)]
    pub worker_capabilities: Vec<String>,
}

/// Response body for `POST /v1/executions/{execution_id}/reclaim`
/// (RFC-024 §3.5).
///
/// The server serializes this struct with a `status` discriminator so
/// consumers can match on structured outcomes without re-parsing a
/// 200-vs-4xx split for business-logic outcomes (mirrors
/// `RotateWaitpointSecretResponse`'s precedent).
#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
#[serde(tag = "status", rename_all = "snake_case")]
pub enum IssueReclaimGrantResponse {
    /// Grant issued. Build a
    /// [`ff_core::contracts::ReclaimGrant`] via
    /// [`Self::into_grant`] and feed it to
    /// [`crate::FlowFabricWorker::claim_from_reclaim_grant`].
    Granted {
        execution_id: String,
        partition_key: ff_core::partition::PartitionKey,
        grant_key: String,
        expires_at_ms: u64,
        lane_id: String,
    },
    /// Execution is not in a reclaimable state (not
    /// `lease_expired_reclaimable` / `lease_revoked`).
    NotReclaimable {
        execution_id: String,
        detail: String,
    },
    /// `max_reclaim_count` exceeded; execution transitioned to
    /// terminal_failed. Consumers stop retrying and surface a
    /// structural failure.
    ReclaimCapExceeded {
        execution_id: String,
        reclaim_count: u32,
    },
}

impl IssueReclaimGrantResponse {
    /// Convert a [`Self::Granted`] response into a typed
    /// [`ff_core::contracts::ReclaimGrant`] for handoff to
    /// [`crate::FlowFabricWorker::claim_from_reclaim_grant`].
    ///
    /// Returns [`SdkError::AdminApi`] when the wire variant is not
    /// `Granted` (consumer asked for a grant but the server replied
    /// with a terminal outcome) or when `execution_id` / `lane_id`
    /// are malformed — the latter signals a drift between server and
    /// SDK, so failing loud prevents silent misrouting.
    pub fn into_grant(self) -> Result<ff_core::contracts::ReclaimGrant, SdkError> {
        match self {
            IssueReclaimGrantResponse::Granted {
                execution_id,
                partition_key,
                grant_key,
                expires_at_ms,
                lane_id,
            } => {
                let eid = ff_core::types::ExecutionId::parse(&execution_id)
                    .map_err(|e| SdkError::AdminApi {
                        status: 200,
                        message: format!(
                            "issue_reclaim_grant: server returned malformed execution_id '{execution_id}': {e}"
                        ),
                        kind: Some("malformed_response".to_owned()),
                        retryable: Some(false),
                        raw_body: String::new(),
                    })?;
                let lane = ff_core::types::LaneId::try_new(lane_id.clone())
                    .map_err(|e| SdkError::AdminApi {
                        status: 200,
                        message: format!(
                            "issue_reclaim_grant: server returned malformed lane_id '{lane_id}': {e}"
                        ),
                        kind: Some("malformed_response".to_owned()),
                        retryable: Some(false),
                        raw_body: String::new(),
                    })?;
                Ok(ff_core::contracts::ReclaimGrant::new(
                    eid,
                    partition_key,
                    grant_key,
                    expires_at_ms,
                    lane,
                ))
            }
            IssueReclaimGrantResponse::NotReclaimable { execution_id, detail } => {
                Err(SdkError::AdminApi {
                    status: 200,
                    message: format!(
                        "issue_reclaim_grant: execution '{execution_id}' not reclaimable: {detail}"
                    ),
                    kind: Some("not_reclaimable".to_owned()),
                    retryable: Some(false),
                    raw_body: String::new(),
                })
            }
            IssueReclaimGrantResponse::ReclaimCapExceeded {
                execution_id,
                reclaim_count,
            } => Err(SdkError::AdminApi {
                status: 200,
                message: format!(
                    "issue_reclaim_grant: execution '{execution_id}' hit reclaim cap ({reclaim_count})"
                ),
                kind: Some("reclaim_cap_exceeded".to_owned()),
                retryable: Some(false),
                raw_body: String::new(),
            }),
        }
    }
}

/// Request body for `POST /v1/admin/rotate-waitpoint-secret`.
///
/// Mirrors `ff_server::api::RotateWaitpointSecretBody` 1:1.
#[derive(Debug, Clone, Serialize)]
pub struct RotateWaitpointSecretRequest {
    /// New key identifier. Non-empty, must not contain `:` (the
    /// server uses `:` as the field separator in the secret hash).
    pub new_kid: String,
    /// Hex-encoded new secret. Even-length, `[0-9a-fA-F]`.
    pub new_secret_hex: String,
}

/// Response body for `POST /v1/admin/rotate-waitpoint-secret`.
///
/// Mirrors `ff_server::server::RotateWaitpointSecretResult` 1:1.
/// The server serializes this struct as-is via `Json(result)`.
#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
pub struct RotateWaitpointSecretResponse {
    /// Count of partitions that accepted the rotation.
    pub rotated: u16,
    /// Partition indices where the rotation failed — operator
    /// should investigate. Rotation is idempotent on the same
    /// `(new_kid, new_secret_hex)` so a retry after the underlying
    /// fault clears converges.
    pub failed: Vec<u16>,
    /// The `new_kid` that was installed as current on every
    /// rotated partition — echoes the request field back for
    /// confirmation.
    pub new_kid: String,
}

/// Server-side error body shape, as emitted by
/// `ff_server::api::ErrorBody`. Kept internal because consumers
/// match on the flattened fields of [`SdkError::AdminApi`].
#[derive(Debug, Clone, Deserialize)]
struct AdminErrorBody {
    error: String,
    #[serde(default)]
    kind: Option<String>,
    #[serde(default)]
    retryable: Option<bool>,
}

/// Request body for `POST /v1/workers/{worker_id}/claim`.
///
/// Mirrors `ff_server::api::ClaimForWorkerBody` 1:1. `worker_id`
/// goes in the URL path (not the body) but is kept on the struct
/// for ergonomics — callers don't juggle a separate arg.
#[derive(Debug, Clone, Serialize)]
pub struct ClaimForWorkerRequest {
    #[serde(skip)]
    pub worker_id: String,
    pub lane_id: String,
    pub worker_instance_id: String,
    #[serde(default)]
    pub capabilities: Vec<String>,
    /// Grant TTL in milliseconds. Server rejects 0 or anything over
    /// 60s (its `CLAIM_GRANT_TTL_MS_MAX`).
    pub grant_ttl_ms: u64,
}

/// Response body for `POST /v1/workers/{worker_id}/claim`.
///
/// Wire shape of `ff_core::contracts::ClaimGrant`. Carries the opaque
/// [`ff_core::partition::PartitionKey`] directly on the wire (issue
/// #91); the SDK reconstructs the core type via [`Self::into_grant`].
#[derive(Debug, Clone, Deserialize)]
pub struct ClaimForWorkerResponse {
    pub execution_id: String,
    pub partition_key: ff_core::partition::PartitionKey,
    pub grant_key: String,
    pub expires_at_ms: u64,
}

impl ClaimForWorkerResponse {
    /// Convert the wire DTO into a typed
    /// [`ff_core::contracts::ClaimGrant`] for handoff to
    /// [`crate::FlowFabricWorker::claim_from_grant`]. Returns
    /// [`SdkError::AdminApi`] on malformed execution_id — a drift
    /// signal that the server and SDK disagree on the wire shape, so
    /// failing loud prevents routing to a ghost partition.
    ///
    /// The `partition_key` itself is not eagerly parsed here: it is
    /// carried opaquely to the `claim_from_grant` hot path, which
    /// parses it there and surfaces a typed error on malformed keys.
    pub fn into_grant(self) -> Result<ff_core::contracts::ClaimGrant, SdkError> {
        let execution_id = ff_core::types::ExecutionId::parse(&self.execution_id)
            .map_err(|e| SdkError::AdminApi {
                status: 200,
                message: format!(
                    "claim_for_worker: server returned malformed execution_id '{}': {e}",
                    self.execution_id
                ),
                kind: Some("malformed_response".to_owned()),
                retryable: Some(false),
                raw_body: String::new(),
            })?;
        Ok(ff_core::contracts::ClaimGrant::new(
            execution_id,
            self.partition_key,
            self.grant_key,
            self.expires_at_ms,
        ))
    }
}

/// Per-partition outcome of a cluster-wide waitpoint HMAC secret
/// rotation. Returned by [`rotate_waitpoint_hmac_secret_all_partitions`]
/// so operators can audit which partitions rotated vs. no-op'd vs. failed.
///
/// The index is the execution-partition index (`0..num_partitions`),
/// matching `{fp:N}` in the keyspace.
#[derive(Debug)]
pub struct PartitionRotationOutcome {
    /// Execution partition index (`0..num_partitions`).
    pub partition: u16,
    /// FCALL outcome on this partition, or the error it raised.
    pub result: Result<RotateWaitpointHmacSecretOutcome, SdkError>,
}

/// Rotate the waitpoint HMAC secret across every execution partition
/// by fanning out the `ff_rotate_waitpoint_hmac_secret` FCALL.
///
/// This is the canonical Rust-side rotation path for direct-Valkey
/// consumers (e.g. cairn-fabric) that cannot route through the
/// `ff-server` admin REST endpoint. Callers who have an HTTP-reachable
/// `ff-server` should prefer [`FlowFabricAdminClient::rotate_waitpoint_secret`] —
/// that path adds a single-writer admission gate, parallel fan-out,
/// structured audit events, and the server's configured grace window.
///
/// # Production rotation recipe
///
/// Operators MUST coordinate so secret rotation **precedes** any
/// waitpoint resolution that will present the new `kid`. The broad
/// sequence:
///
/// 1. Pick a fresh `new_kid` (must NOT contain `:` — the server uses
///    `:` as the field separator in the secret hash).
/// 2. Call this helper with the previous `kid`'s grace window
///    (`grace_ms` — the duration during which tokens signed by the
///    outgoing secret remain valid).
/// 3. Only after this call returns with all partitions `Ok(_)` (either
///    `Rotated` or `Noop`), begin signing new tokens with `new_kid`.
/// 4. Retain the previous secret in the keystore until the grace
///    window elapses — the FCALL handles GC of expired kids on every
///    rotation, so just don't rotate again before the grace window.
///
/// See RFC-004 §rotation for the full 4-key HSET + `previous_expires_at`
/// dance the FCALL implements server-side.
///
/// # Idempotency
///
/// Each partition FCALL is idempotent on the same `(new_kid,
/// new_secret_hex)` pair: a replay with identical args returns
/// `RotateWaitpointHmacSecretOutcome::Noop`. A same-kid-different-secret
/// replay surfaces as a per-partition `SdkError` (wrapping
/// `ScriptError::RotationConflict`) — pick a fresh `new_kid` to recover.
///
/// # Error semantics
///
/// A per-partition FCALL failure (transport fault, rotation conflict,
/// etc.) is recorded on that partition's [`PartitionRotationOutcome`]
/// and fan-out **continues** — the contract matches the server's
/// `rotate_waitpoint_secret` (partial success is allowed, operators
/// retry on the failed partition subset). Returning `Vec<_>` (not
/// `Result<Vec<_>, _>`) is deliberate: every whole-call invariant is
/// enforced by the underlying FCALL on each partition (kid non-empty,
/// no `:`, even-length hex, etc.), so the aggregate has nothing left
/// to reject at the Rust boundary. Callers decide how to treat partial
/// failures (fail loud / retry the subset / record metrics).
///
/// # Concurrency + performance
///
/// Sequential (one partition at a time) to keep the helper dependency-
/// free: no `futures::stream` / tokio-specific primitives on the caller
/// path. For a cluster with N partitions and per-partition RTT R, the
/// total duration is ~N*R. Consumers needing parallel fan-out should
/// wrap this with `FuturesUnordered` themselves, or use the server
/// admin endpoint (which fans out with bounded concurrency = 16).
///
/// # Test harness
///
/// The `ff-test::fixtures::TestCluster::rotate_waitpoint_hmac_secret`
/// method is a thin wrapper around this helper — integration tests and
/// production code exercise the same code path.
///
/// # Example
///
/// ```rust,ignore
/// use ff_sdk::admin::rotate_waitpoint_hmac_secret_all_partitions;
///
/// let results = rotate_waitpoint_hmac_secret_all_partitions(
///     &client,
///     partition_config.num_flow_partitions,
///     "kid-2026-04-22",
///     "deadbeef...64-hex-chars...",
///     60_000,
/// )
/// .await?;
///
/// for entry in &results {
///     match &entry.result {
///         Ok(outcome) => tracing::info!(partition = entry.partition, ?outcome, "rotated"),
///         Err(e) => tracing::error!(partition = entry.partition, %e, "rotation failed"),
///     }
/// }
/// ```
// v0.12 PR-6: the `admin` module is ungated at module level so
// consumers under `--no-default-features --features sqlite` can reach
// the HTTP admin client surface. This helper is the one remaining
// Valkey-typed item in the module (takes a `&ferriskey::Client` and
// fans out `ff_rotate_waitpoint_hmac_secret` FCALLs), so it stays
// `valkey-default`-gated. See `lib.rs` PR-6 comment for the Option 1
// / Option 2 decision.
#[cfg(feature = "valkey-default")]
pub async fn rotate_waitpoint_hmac_secret_all_partitions(
    client: &ferriskey::Client,
    num_partitions: u16,
    new_kid: &str,
    new_secret_hex: &str,
    grace_ms: u64,
) -> Vec<PartitionRotationOutcome> {
    // Hoisted out of the loop — `ff_rotate_waitpoint_hmac_secret` only
    // borrows the args, so every partition can reuse the same struct.
    // Avoids N × 2 string clones on the hot fan-out path.
    let args = RotateWaitpointHmacSecretArgs {
        new_kid: new_kid.to_owned(),
        new_secret_hex: new_secret_hex.to_owned(),
        grace_ms,
    };
    let mut out = Vec::with_capacity(num_partitions as usize);
    for index in 0..num_partitions {
        let partition = Partition {
            family: PartitionFamily::Execution,
            index,
        };
        let idx = IndexKeys::new(&partition);
        let result = ff_script::functions::suspension::ff_rotate_waitpoint_hmac_secret(
            client, &idx, &args,
        )
        .await
        .map_err(SdkError::from);
        out.push(PartitionRotationOutcome {
            partition: index,
            result,
        });
    }
    out
}

/// Trim trailing slashes from a base URL so `format!("{base}/v1/...")`
/// never produces `https://host//v1/...`. Mirror of
/// media-pipeline's pattern.
fn normalize_base_url(mut url: String) -> String {
    while url.ends_with('/') {
        url.pop();
    }
    url
}

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

    #[test]
    fn base_url_strips_trailing_slash() {
        assert_eq!(normalize_base_url("http://x".into()), "http://x");
        assert_eq!(normalize_base_url("http://x/".into()), "http://x");
        assert_eq!(normalize_base_url("http://x///".into()), "http://x");
    }

    #[test]
    fn with_token_rejects_bad_header_chars() {
        // Raw newline in the token would split the Authorization
        // header — must fail loudly at construction.
        let err = FlowFabricAdminClient::with_token("http://x", "tok\nevil").unwrap_err();
        assert!(
            matches!(err, SdkError::Config { .. }),
            "got: {err:?}"
        );
    }

    #[test]
    fn with_token_rejects_empty_or_whitespace() {
        // Exact shell footgun: FF_ADMIN_TOKEN="" expands to "".
        // Fail loudly at construction instead of shipping a client
        // that silently 401s on first request.
        for s in ["", " ", "\t\n ", "   "] {
            let err = FlowFabricAdminClient::with_token("http://x", s)
                .unwrap_err();
            assert!(
                matches!(&err, SdkError::Config { field: Some(f), .. } if f == "bearer_token"),
                "token {s:?} should return Config with field=bearer_token; got: {err:?}"
            );
        }
    }

    #[test]
    fn admin_error_body_deserialises_optional_fields() {
        // `kind` + `retryable` absent (the usual shape for 400s).
        let b: AdminErrorBody = serde_json::from_str(r#"{"error":"bad new_kid"}"#).unwrap();
        assert_eq!(b.error, "bad new_kid");
        assert!(b.kind.is_none());
        assert!(b.retryable.is_none());

        // `kind` + `retryable` present (500 ValkeyError shape).
        let b: AdminErrorBody = serde_json::from_str(
            r#"{"error":"valkey: timed out","kind":"IoError","retryable":true}"#,
        )
        .unwrap();
        assert_eq!(b.error, "valkey: timed out");
        assert_eq!(b.kind.as_deref(), Some("IoError"));
        assert_eq!(b.retryable, Some(true));
    }

    #[test]
    fn rotate_response_deserialises_server_shape() {
        // Exact shape the server emits.
        let raw = r#"{
            "rotated": 3,
            "failed": [4, 5],
            "new_kid": "kid-2026-04-18"
        }"#;
        let r: RotateWaitpointSecretResponse = serde_json::from_str(raw).unwrap();
        assert_eq!(r.rotated, 3);
        assert_eq!(r.failed, vec![4, 5]);
        assert_eq!(r.new_kid, "kid-2026-04-18");
    }

    // ── ClaimForWorkerResponse::into_grant ──

    fn sample_claim_response(partition_key: &str) -> ClaimForWorkerResponse {
        ClaimForWorkerResponse {
            execution_id: "{fp:5}:11111111-1111-1111-1111-111111111111".to_owned(),
            partition_key: serde_json::from_str(
                &serde_json::to_string(partition_key).unwrap(),
            )
            .unwrap(),
            grant_key: "ff:exec:{fp:5}:11111111-1111-1111-1111-111111111111:claim_grant".to_owned(),
            expires_at_ms: 1_700_000_000_000,
        }
    }

    #[test]
    fn into_grant_preserves_all_known_partition_key_shapes() {
        // Post-#91: families collapse into opaque PartitionKey literals.
        // Flow and Execution both produce "{fp:N}"; Budget is "{b:N}";
        // Quota is "{q:N}". The DTO preserves the wire string as-is;
        // into_grant hands it opaquely to the core type.
        for key_str in ["{fp:5}", "{b:5}", "{q:5}"] {
            let g = sample_claim_response(key_str).into_grant().unwrap_or_else(|e| {
                panic!("key {key_str} should parse: {e:?}")
            });
            assert_eq!(g.partition_key.as_str(), key_str);
            assert_eq!(g.expires_at_ms, 1_700_000_000_000);
        }
    }

    #[test]
    fn into_grant_preserves_opaque_partition_key() {
        // The SDK does NOT eagerly parse the partition_key on the
        // admin boundary — malformed keys are caught at the
        // claim_from_grant hot path where the typed Partition is
        // actually needed. This test pins the opacity contract.
        let resp = sample_claim_response("{zz:0}");
        let g = resp.into_grant().expect("SDK must not parse partition_key");
        assert_eq!(g.partition_key.as_str(), "{zz:0}");
        // Parsing surfaces the error explicitly.
        assert!(g.partition().is_err());
    }

    #[test]
    fn into_grant_rejects_malformed_execution_id() {
        let mut resp = sample_claim_response("{fp:5}");
        resp.execution_id = "not-a-valid-eid".to_owned();
        let err = resp.into_grant().unwrap_err();
        match err {
            SdkError::AdminApi { message, kind, .. } => {
                assert!(message.contains("malformed execution_id"),
                    "msg: {message}");
                assert_eq!(kind.as_deref(), Some("malformed_response"));
            }
            other => panic!("expected AdminApi, got {other:?}"),
        }
    }

    // ── ClaimForWorkerResponse wire shape (issue #91) ──

    // `rotate_waitpoint_hmac_secret_all_partitions` exercise coverage
    // lives in `ff-test` — the integration test harness in
    // `crates/ff-test/tests/waitpoint_hmac_rotation_fcall.rs` and
    // `waitpoint_tokens.rs` calls through the function via the
    // `TestCluster::rotate_waitpoint_hmac_secret` fixture, which is
    // now a thin delegator. A pure unit test here would require a
    // mock `ferriskey::Client` (ferriskey's `Client` performs a live
    // RESP handshake on `ClientBuilder::build`, so a local TCP
    // listener alone isn't sufficient) — expensive to construct for
    // one-line iteration-count coverage.

    #[test]
    fn read_waitpoint_token_url_percent_encodes_path_segments() {
        // The execution id carries `{fp:N}:` literal punctuation;
        // a naïve `format!` splice would ship those chars unencoded
        // and the server would match the wrong route. Pin that the
        // reqwest URL builder percent-encodes each segment.
        use ff_core::types::{ExecutionId, WaitpointId};

        let mut url = reqwest::Url::parse("http://x").unwrap();
        let execution_id = ExecutionId::parse(
            "{fp:7}:11111111-1111-1111-1111-111111111111",
        )
        .unwrap();
        let waitpoint_id = WaitpointId::parse("22222222-2222-2222-2222-222222222222")
            .unwrap();
        url.path_segments_mut()
            .unwrap()
            .extend(&[
                "v1",
                "executions",
                execution_id.as_str(),
                "waitpoints",
                &waitpoint_id.to_string(),
                "token",
            ]);
        assert_eq!(
            url.as_str(),
            "http://x/v1/executions/%7Bfp:7%7D:11111111-1111-1111-1111-111111111111\
             /waitpoints/22222222-2222-2222-2222-222222222222/token"
        );
    }

    #[test]
    fn claim_for_worker_response_deserialises_opaque_partition_key() {
        // Exact shape the server emits post-#91.
        let raw = r#"{
            "execution_id": "{fp:7}:11111111-1111-1111-1111-111111111111",
            "partition_key": "{fp:7}",
            "grant_key": "ff:exec:{fp:7}:11111111-1111-1111-1111-111111111111:claim_grant",
            "expires_at_ms": 1700000000000
        }"#;
        let r: ClaimForWorkerResponse = serde_json::from_str(raw).unwrap();
        assert_eq!(r.partition_key.as_str(), "{fp:7}");
        assert_eq!(r.expires_at_ms, 1_700_000_000_000);
    }
}