polyc-payments 2026.9.0

//! Control-plane payment proxy core.
//!
//! When outbound `paid_fetch` signing moves off the untrusted sandbox, the
//! sandbox sends an intent (a `PaidFetchRequest`) up the harness stream and the
//! trusted control plane fulfills it here. This module is the **chokepoint**:
//! it applies, in order, the spend-control and safety layers that the sandbox
//! must not be trusted to apply itself, then performs the 402-gated fetch with a
//! keychain (or gated raw) signer and returns a
//! [`ProxyOutcome`](crate::proxy::ProxyOutcome).
//!
//! The API is **typed**: the caller parses its tool's argument JSON and hands
//! over the destination `url` and optional `max_spend` (derived from the
//! APPROVED args), and every refusal comes back as a machine-readable
//! [`RejectReason`](crate::proxy::RejectReason) — the reader-facing wording
//! lives with the caller, beside
//! the rest of its surface copy, never in this foundation crate.
//!
//! Enforcement order (matches the design's spend-control stack):
//! 1. **Approval binding** — the request's `args_json` must value-match the args
//!    the human approved (do not trust the sandbox to send what was approved).
//! 2. **SSRF guard** — the fetch URL must not target an internal/metadata host.
//! 3. **Pre-authorization** — the intended per-call spend must fit the
//!    granted per-transaction / per-session ceiling (the pre-signing
//!    spend cap, a [`SpendCap`](polyc_spend_policy::presign::SpendCap)) AND, when a signed
//!    AP2-style mandate chain is presented and a trusted issuer key is
//!    configured, the chain's amount/merchant/args authorization
//!    ([`VerifiedMandateChain`](polyc_crypto::mandate::VerifiedMandateChain)) — both
//!    rejected *before* the budget is reserved and before any credential is
//!    minted. Mandates are additive and off by default: with no presented
//!    chain (or no `TEMPO_MANDATE_ISSUER_PUBKEY` configured), only the cap
//!    applies, exactly as before.
//! 4. **Aggregate budget** — read the conversation's durable floor and reserve
//!    the per-call ceiling against it *before* signing; resolve the reservation
//!    **at the moment of settlement**, in the same uninterrupted step that mints
//!    the credential, so a cancelled turn cannot land between a real charge and
//!    the record of it. An authority that cannot be reached refuses the payment
//!    rather than reporting a zero floor.
//! 5. **Per-call cap + chain guard** — applied by the `CappedProvider` the
//!    caller wraps the signer in (rejects an over-cap or wrong-chain challenge
//!    before signing).
//! 6. **Onchain key limit + expiry** — the keychain access key itself (outer
//!    net), enforced by the chain.
//!
//! [`fulfill_with`](crate::proxy::fulfill_with) is the injectable, unit-tested
//! core (provider supplied); [`fulfill`](crate::proxy::fulfill) is the
//! production wrapper that resolves the signer from
//! `PaymentsConfig` and wraps it in a `CappedProvider`.
//!
//! ## One call, one charge
//!
//! **A call gets one charge against the conversation's budget. A merchant that
//! wants more re-challenges on a new call.** This is a decision, not a
//! consequence of ordering, so it is stated here rather than left to be
//! rediscovered.
//!
//! The 402 exchange is willing to pay several *distinct* challenges in a single
//! call: a merchant that answers the paid retry with a new challenge (an
//! incremental charge) gets paid again, up to the exchange's retry limit. This
//! proxy refuses that. One call reserves once, and the reservation the
//! settlement point resolves is single-use, so the second commit comes back
//! [`AlreadyResolved`](polyc_spend_policy::budget::CommitError::AlreadyResolved)
//! and the payment is refused with the credential still in hand — nothing is
//! sent for it.
//!
//! The cost is real and accepted: a merchant charging incrementally hard-fails
//! after its first payment, having taken money and served a 402. The alternative
//! is worse. The aggregate budget is the only ceiling that spans calls, and
//! paying a second challenge the reservation cannot account for moves money the
//! budget never counts — one call spending several times its own cap while the
//! budget records a single charge. Refusing keeps the two in step, and refusing
//! *before* the credential leaves means the refusal itself costs nothing.
//!
//! A new call is the supported path for the merchant's second charge: it re-runs
//! the approval binding and reserves budget again, so the spend is authorized
//! and counted like any other. `refuses_a_second_payment_in_one_call` pins the
//! behavior.

use std::sync::Arc;

use mpp::client::PaymentProvider;
use mpp::{MppError, PaymentChallenge, PaymentCredential};
use polyc_crypto::canon::canon_args;
use polyc_crypto::mandate::{self, MandateChain, MandateError, VerifiedMandateChain};
use polyc_egress::egress::{CapMode, pinned_http_client, pinned_http_client_with_timeout};
use polyc_egress::ssrf::{self, SsrfError, SsrfPolicy};
use polyc_payments_client::outbound::{
    CappedProvider, OutboundRequest, PaymentError, paid_request_with_client,
};
use polyc_payments_client::resolver::{KeySource, PayerKind};
use polyc_spend_policy::budget::{
    CommitError, ConversationSpend, ReserveRefused, Settlement, SharedReservation,
};
use polyc_spend_policy::presign::{CapExceeded, SpendCap};

use crate::config::PaymentsConfig;

/// Wraps a [`PaymentProvider`] so the call's reservation is resolved **at the
/// settlement point** — inside `pay`, in the same uninterrupted step that mints
/// the credential.
///
/// # Why the resolution lives here
///
/// A credential only costs money once it reaches the merchant, and the only way
/// out of this `pay` is the `Ok` below. Resolving before that return means a
/// charge cannot exist without the reservation already carrying it, and a
/// resolution that did not run means a credential that never escaped this
/// process. A turn deadline that fires anywhere in the exchange therefore
/// cannot lose a real charge nor invent one — there is no interval between the
/// charge and its record for the cancellation to land in. See
/// [`SharedReservation`] for the full argument.
///
/// The durable write is deliberately NOT here. A durable write is an await, and
/// an await between the credential and the return is exactly the window this
/// ordering closes. The caller performs it afterwards, keyed by the reservation
/// identity so the normal return and a cancellation cannot both count the same
/// charge; see [`SettlementSink`].
///
/// A failed resolution refuses the payment, discarding the credential this
/// function is holding. That is the fail-closed direction: nothing was sent, so
/// no money moves.
///
/// The reservation records a real settlement even when the upstream returns no
/// parseable `Payment-Receipt` (a paid service that omits or malforms the
/// receipt must still consume budget). A free (non-402) fetch never calls `pay`,
/// so its reservation goes back untouched.
///
/// # A second settlement in the same call is refused
///
/// The reservation resolves once, so a merchant that answers the paid retry with
/// another challenge gets its second payment refused here — before that
/// credential leaves. See the module docs' "one call, one charge" section for
/// the decision and what it costs.
///
/// # A zero-amount challenge still consumes the call's one charge
///
/// A challenge for zero settles: the provider returns a proof credential and no
/// transfer is made. This decorator treats it as a settlement like any other. It
/// records zero, so the caller attributes `amount_base_units: Some(0)` and
/// commits a charge that moved nothing. It also spends the call's single charge,
/// so a real challenge arriving after it in the same call is refused by the rule
/// above. Harmless on the ceiling — the money figure is honest at zero — but the
/// empty commit and the consumed charge are both real, and neither is worth new
/// machinery until a merchant actually charges this way.
struct PaymentObserver<'a, P> {
    inner: &'a P,
    reservation: SharedReservation,
}

// Manual `Clone` (the provider is a shared reference and the reservation is
// `Arc`-backed, so no `P: Clone` bound is needed). It exists only to satisfy
// `PaymentProvider`'s `Clone` supertrait: the 402 exchange clones the REQUEST
// BUILDER for its authed retry and holds the provider by reference throughout,
// so this decorator is never actually cloned on the payment path. Were it
// cloned, the clone would share the same reservation, so one call still gets one
// charge.
impl<P> Clone for PaymentObserver<'_, P> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner,
            reservation: self.reservation.clone(),
        }
    }
}

impl<P: PaymentProvider> PaymentProvider for PaymentObserver<'_, P> {
    fn supports(&self, method: &str, intent: &str) -> bool {
        self.inner.supports(method, intent)
    }
    async fn pay(&self, challenge: &PaymentChallenge) -> Result<PaymentCredential, MppError> {
        // Resolve the reservation ONLY after the wrapped provider actually
        // produces a credential. The inner provider is the `CappedProvider`,
        // which rejects wrong-chain / unknown-currency / over-cap / unparseable
        // challenges *before* signing — those must NOT consume budget or emit a
        // spend event (no credential was created).
        let credential = self.inner.pay(challenge).await?;
        // Capture the charged amount from the challenge. The inner CappedProvider
        // already parsed/validated this exact amount before signing, so on success
        // it decodes.
        let charged = polyc_payments_client::outbound::challenge_amount_base_units(challenge)
            .ok()
            .map(|amount| u128::try_from(amount).unwrap_or(u128::MAX));
        // THE CANCELLATION-SAFE ORDER. Everything from here to the `Ok` below is
        // synchronous: no `.await` runs, so no cancellation point exists between
        // the reservation resolving and the credential leaving this function.
        // Reversing these two — returning first and resolving when control gets
        // back to `fulfill_with` — is exactly the window a turn deadline used to
        // fall into.
        if charged.is_none() {
            // A real settlement whose figure could not be read must never round
            // to zero, so the reserved ceiling stands in for it below.
            tracing::warn!(
                "outbound payment settled but charged amount was not captured; \
                 recording the full reserved cap"
            );
        }
        let resolved = charged.map_or_else(
            || self.reservation.commit_full(),
            |amount| self.reservation.commit_actual(amount),
        );
        if let Err(e) = resolved {
            // Nothing has been sent, so refusing here spends nothing. Fail closed
            // rather than hand over a credential no reservation accounts for.
            //
            // The two failures are different facts and must not be worded alike.
            // `AlreadyResolved` is the one-call-one-charge rule doing its job —
            // the record is healthy and this call has simply already paid — while
            // `Unavailable` really is the record being unwritable. Matching here
            // rather than interpolating `{e}` is what keeps a working ceiling from
            // being reported as a broken one.
            return Err(MppError::verification_failed(match e {
                CommitError::AlreadyResolved => {
                    "this request already paid once, and each approved request pays \
                     at most once, so the site\'s follow-up charge was not signed and \
                     no money moved for it"
                }
                CommitError::Unavailable => {
                    "this conversation\'s spending record could not be updated, so the \
                     payment was not completed and no money moved"
                }
            }));
        }
        Ok(credential)
    }
}

/// Cancellation-safety net for the DURABLE settlement record (#168b).
///
/// The reservation needs no net: it resolves at the settlement point itself,
/// inside the provider decorator that mints the credential, so no window exists
/// between a real charge and the record of it (#168a — see
/// [`SharedReservation`]). The durable write is different, because a durable
/// write is an await and an await there would reopen exactly that window. The
/// CALLER performs it after [`fulfill_with`] returns, and a drop in between
/// resolves the reservation yet never reaches that write.
///
/// This seam closes the remaining window. [`fulfill_with`] arms a
/// [`SettlementNet`] over the whole call and hands it back to the caller **still
/// armed**, so the net spans the caller's durable writes as well as its own
/// region. On a drop anywhere before those writes land, the net spawns this sink
/// to record the spend durably.
///
/// # One record, whichever path writes it
///
/// The two paths write the SAME record, keyed by the call's reservation
/// identity, so a second attempt is a no-op rather than a second charge. That
/// is what makes the arm/disarm seam safe to rely on rather than merely
/// convenient: were both to fire — a future edit, a torn disarm — the
/// conversation's committed total still moves once.
///
/// This is the shape C3.1's constraint required. Adding a durable commit
/// *beside* an independently-keyed receipt would put a cancelled paid turn's
/// charge into the same floor twice, permanently burning that headroom on every
/// cancelled paid turn. Making them one keyed record is what removes the
/// possibility rather than the occasion.
#[async_trait::async_trait]
pub trait SettlementSink: Send + Sync {
    /// Durably record a settled payment of `amount` base units against the
    /// call's reservation. Invoked at most once per call, from
    /// [`SettlementNet`]'s `Drop`. Best-effort: the implementation swallows
    /// durable-write errors (the onchain key cap is the maximum-loss backstop),
    /// and the caller cannot await it (the turn future may be being dropped).
    async fn persist_settlement(&self, amount: u128);
}

/// The armed half of a [`SettlementNet`]: everything a `Drop`-time write needs.
struct ArmedNet {
    sink: Arc<dyn SettlementSink>,
    reservation: SharedReservation,
    handle: tokio::runtime::Handle,
}

/// Armed net that turns a lost settlement write into a durable record (#168b).
///
/// [`fulfill_with`] arms one before its first `.await` and returns it to the
/// caller still armed. It shares the call's [`SharedReservation`]; on `Drop`,
/// when that reservation resolved to a charge and the net is still armed, it
/// spawns [`SettlementSink::persist_settlement`] so the spend is recorded even
/// though no `.await` can run during a drop.
///
/// # The net spans the caller's writes, not only this crate's
///
/// The durable record is the CALLER's write, and it is an `.await` that happens
/// after this function returns. A net that disarmed on the normal return would
/// therefore cover everything except the one write it exists for: a turn
/// deadline landing on that `.await` would resolve the reservation, drop the
/// future, and reach nothing. So the net is handed out armed, and
/// [`Self::disarm`] means one specific thing — *the durable record is already
/// on the ledger*. Anything else leaves it armed and the write happens: a drop
/// mid-commit, a commit that failed, a caller that never disarmed at all.
///
/// Firing when the caller already wrote costs nothing. Both paths write the same
/// keyed record, so the second arrival is a replay rather than a second charge.
pub struct SettlementNet {
    /// `Some` while the net still owes a write. `None` on a call with no durable
    /// log configured, and after [`Self::disarm`].
    armed: Option<ArmedNet>,
}

impl SettlementNet {
    /// Arms a net sharing `reservation`, capturing the current runtime handle so
    /// a `Drop`-time persist can be spawned (a drop cannot await). Called from
    /// inside [`fulfill_with`], which always runs on a runtime. A `None` sink —
    /// no durable log — produces an inert net.
    fn arm(sink: Option<Arc<dyn SettlementSink>>, reservation: SharedReservation) -> Self {
        Self {
            armed: sink.map(|sink| ArmedNet {
                sink,
                reservation,
                handle: tokio::runtime::Handle::current(),
            }),
        }
    }

    /// An inert net, for a call refused before the payment exchange began.
    ///
    /// Nothing settled, so there is nothing for a net to record.
    const fn inert() -> Self {
        Self { armed: None }
    }

    /// Declares the settled payment's durable record already written.
    ///
    /// Call this only once the keyed spend commit and the settlement receipt
    /// have both landed. Leaving a net armed writes a record that is already
    /// there, which the keyed identity absorbs; disarming one whose write never
    /// landed loses a real charge.
    pub fn disarm(&mut self) {
        self.armed = None;
    }
}

impl Drop for SettlementNet {
    fn drop(&mut self) {
        // Only a settled payment whose record has not landed is written here:
        // `disarm` clears the net once the caller's own write succeeded, and a
        // call that never charged has nothing to record.
        let Some(armed) = self.armed.take() else {
            return;
        };
        let Some(Settlement::Charged(amount)) = armed.reservation.settlement() else {
            return;
        };
        let sink = armed.sink;
        // The turn future may be being dropped, so we cannot await — spawn a
        // detached persist. The runtime outlives the cancelled turn (only the turn
        // future was dropped, not the process), so it completes; on the control
        // plane's multi-thread runtime it runs immediately on another worker.
        drop(armed.handle.spawn(async move {
            sink.persist_settlement(amount).await;
        }));
    }
}

/// Conservative built-in per-call ceiling (dollars) when none is configured.
/// Mirrors `paid_fetch`'s floor so an outbound payment is never uncapped.
///
/// `pub(crate)` so [`crate::config::PaymentsConfig::default_max_spend_base_units`]
/// can report the SAME effective figure the proxy actually enforces, rather
/// than a second hand-copied literal that could drift from this one.
pub(crate) const DEFAULT_CAP: &str = "0.10";

/// Settlement facts lifted from the `Payment-Receipt`, for the durable
/// `outbound_payment_receipt` event.
///
/// Only fields the receipt actually carries — notably **not** the amount (a
/// Tempo receipt does not carry one), so the economic record stores the onchain
/// `reference` + `currency` rather than a fabricated figure.
#[derive(Debug, Default, Clone)]
pub struct ProxyReceipt {
    /// Onchain settlement reference (tx hash) or native id.
    pub reference: String,
    /// Payment method (e.g. `"tempo"`).
    pub method: String,
    /// Settlement timestamp.
    pub timestamp: String,
}

/// Deliberately drops `mpp::Receipt`'s `status` and `external_id` — this is a
/// subset projection for the durable `outbound_payment_receipt` event, not a
/// full mirror (see the struct doc above for why `amount` is absent too).
impl From<&mpp::Receipt> for ProxyReceipt {
    fn from(r: &mpp::Receipt) -> Self {
        Self {
            reference: r.reference.clone(),
            method: r.method.as_str().to_string(),
            timestamp: r.timestamp.clone(),
        }
    }
}

/// A payment that **settled** during the exchange: who paid and how much.
///
/// Present on both outcome variants — a payment can settle and the call still
/// fail afterwards (a post-pay body-read failure), and that spend must never
/// read as a clean failure.
#[derive(Debug, Default, Clone)]
pub struct PaidAttribution {
    /// The **charged base-unit amount** captured from the challenge, when it
    /// could be decoded. The token + decimals context is supplied by the caller
    /// (from the configured `TEMPO_CURRENCY` / `TEMPO_CURRENCY_DECIMALS`).
    pub amount_base_units: Option<u128>,
    /// The onchain account (0x…) that actually settled the payment, when
    /// known — filled by [`fulfill`], which knows the resolved signer. The
    /// accurate payer to attribute the spend to.
    pub paying_account: Option<String>,
    /// The settled amount rendered for a person to read, with its currency
    /// symbol (e.g. `"0.10 USD"`) — filled by [`fulfill`] from the charged
    /// amount at the configured decimals.
    pub paid_amount: Option<String>,
    /// Whose account paid — the caller's linked wallet or the deployment's own
    /// signer (see [`PayerKind`]) — filled by [`fulfill`] from the key source.
    pub payer_kind: Option<PayerKind>,
}

/// Why the proxy refused (or failed) a fetch — machine-readable, so the caller
/// words the reader-facing message itself, beside the rest of its copy.
///
/// The `Display` form is a terse diagnostic for logs and tests; it is NOT the
/// text to show a reader or the model.
#[derive(Debug, thiserror::Error)]
pub enum RejectReason {
    /// The sandbox-sent args do not value-match the human-approved args.
    #[error("args do not match the approved tool call")]
    ApprovalMismatch,
    /// The SSRF guard refused the destination.
    #[error("blocked destination: {0}")]
    BlockedDestination(SsrfError),
    /// The guarded outbound client could not be built (fatal TLS
    /// misconfiguration) — fail closed rather than fetch unguarded.
    #[error("outbound client build failed")]
    ClientBuild,
    /// The destination host is not on the configured paid-host allowlist.
    #[error("destination host is not on the configured allowlist")]
    HostNotAllowed,
    /// The destination host is on the deployment allowlist (or none is
    /// configured) but is not on the CALLER's own persona spend-policy host
    /// allowlist (`#514`/`#517`) — a distinct reason from [`Self::HostNotAllowed`]
    /// so the caller words the two differently (a deployment-wide restriction
    /// vs. a policy the caller's own admin set on their wallet).
    #[error("destination host is not on this wallet's allowed hosts")]
    PersonaHostNotAllowed,
    /// The per-call ceiling exceeds the pre-authorized spend cap.
    #[error("over pre-authorized spend cap: {0}")]
    OverSpendCap(CapExceeded),
    /// A presented mandate chain refused the payment (or failed to verify).
    #[error("mandate refused: {0}")]
    MandateRefused(MandateError),
    /// A mandate was presented but the destination has no parseable host to
    /// bind it to — a mandate authorizes nothing there (fail closed).
    #[error("mandate refused: destination has no host")]
    MandateHostUnknown,
    /// The reservation would push the conversation over its aggregate budget.
    #[error("over conversation budget: {0}")]
    OverBudget(polyc_spend_policy::budget::BudgetExceeded),
    /// The conversation already holds as many payments in flight as it can.
    ///
    /// Its own class rather than a shade of [`Self::OverBudget`]: nothing about
    /// the conversation's ceiling is exhausted, so a message quoting a spending
    /// figure would name a number that is not the reason.
    #[error("this conversation already has as many payments in flight as it can hold")]
    TooManyPaymentsInFlight,
    /// This approved call already holds budget for a payment that has not
    /// finished.
    ///
    /// Also its own class, and for the same reason: one approved call holds
    /// under one identity, so this is that call arriving twice rather than a
    /// conversation out of budget.
    #[error("this request is already holding budget for a payment that has not finished")]
    PaymentAlreadyInFlight,
    /// The conversation's durable spending record could not be reached, so the
    /// aggregate ceiling could not be applied.
    ///
    /// Its own class rather than a shade of [`Self::OverBudget`]: a conversation
    /// at its ceiling and an authority that cannot say are different facts, and
    /// the reader-facing wording for the two is different. Both refuse the
    /// payment — reporting a zero floor instead would hand the conversation its
    /// whole ceiling again every time the authority blinked.
    #[error("the conversation's spending record could not be reached")]
    SpendUnreachable,
    /// No settlement currency is configured (`TEMPO_CURRENCY`), so the
    /// per-call cap cannot be enforced — refuse rather than sign blind.
    #[error("no settlement currency configured")]
    MissingCurrency,
    /// A supplied `max_spend` (agent-authored or the operator's
    /// `TEMPO_MAX_SPEND`) failed to parse into base units — malformed, or a
    /// magnitude overflowing `u128`. Refuse outright rather than silently
    /// falling back to a wider, unintended cap.
    #[error("max_spend could not be parsed into base units")]
    InvalidMaxSpend,
    /// The outbound signer could not be resolved.
    #[error("payment backend unavailable: {0}")]
    BackendUnavailable(crate::config::PaymentsConfigError),
    /// The (non-paying) upstream GET failed at the transport level.
    #[error("upstream fetch failed")]
    FetchFailed,
    /// The payment layer failed — see [`PaymentError`] for the exact class
    /// (wrong chain, over cap, refused credential, transport…).
    #[error(transparent)]
    Payment(PaymentError),
}

/// The result of a proxied fetch: the upstream response, or a rejection.
///
/// A settled payment ([`PaidAttribution`]) can ride EITHER variant — money
/// that left the wallet is reported even when the call failed afterwards.
#[derive(Debug)]
pub enum ProxyOutcome {
    /// The upstream responded (paid or free).
    Fetched {
        /// HTTP status of the final (post-payment) response.
        status: u16,
        /// Response body (capped).
        body: String,
        /// Raw `Payment-Receipt` header verbatim, when the server returned one.
        receipt_header: Option<String>,
        /// Parsed settlement facts, when the server returned a receipt — for
        /// the `outbound_payment_receipt` event.
        receipt: Option<ProxyReceipt>,
        /// A settled payment's attribution; `None` for a free (non-402) fetch.
        paid: Option<PaidAttribution>,
    },
    /// The fetch was refused or failed.
    Rejected {
        /// The machine-readable refusal/failure class.
        reason: RejectReason,
        /// A settled payment's attribution, when a payment settled BEFORE the
        /// failure (settle-then-fail must not read as a clean failure).
        paid: Option<PaidAttribution>,
    },
}

impl ProxyOutcome {
    const fn rejected(reason: RejectReason) -> Self {
        Self::Rejected { reason, paid: None }
    }

    /// Whether this is a successful fetch (vs a rejection/failure).
    #[must_use]
    pub const fn is_ok(&self) -> bool {
        matches!(self, Self::Fetched { .. })
    }

    /// The rejection reason, when the fetch was refused or failed.
    #[must_use]
    pub const fn reason(&self) -> Option<&RejectReason> {
        match self {
            Self::Fetched { .. } => None,
            Self::Rejected { reason, .. } => Some(reason),
        }
    }

    /// The settled payment's attribution, on either variant, when one settled.
    #[must_use]
    pub const fn paid(&self) -> Option<&PaidAttribution> {
        match self {
            Self::Fetched { paid, .. } | Self::Rejected { paid, .. } => paid.as_ref(),
        }
    }

    /// The upstream HTTP status, on success.
    #[must_use]
    pub const fn status(&self) -> Option<u16> {
        match self {
            Self::Fetched { status, .. } => Some(*status),
            Self::Rejected { .. } => None,
        }
    }

    /// The upstream body, on success.
    #[must_use]
    pub fn body(&self) -> Option<&str> {
        match self {
            Self::Fetched { body, .. } => Some(body),
            Self::Rejected { .. } => None,
        }
    }

    /// The parsed settlement receipt, on a success that carried one.
    #[must_use]
    pub const fn receipt(&self) -> Option<&ProxyReceipt> {
        match self {
            Self::Fetched { receipt, .. } => receipt.as_ref(),
            Self::Rejected { .. } => None,
        }
    }

    /// The raw `Payment-Receipt` header, on a success that carried one.
    #[must_use]
    pub fn receipt_header(&self) -> Option<&str> {
        match self {
            Self::Fetched { receipt_header, .. } => receipt_header.as_deref(),
            Self::Rejected { .. } => None,
        }
    }

    /// Mutable access to the settled-payment attribution, for [`fulfill`] to
    /// fill payer identity after the exchange.
    const fn paid_mut(&mut self) -> Option<&mut PaidAttribution> {
        match self {
            Self::Fetched { paid, .. } | Self::Rejected { paid, .. } => paid.as_mut(),
        }
    }
}

/// Whether `url`'s host is permitted by an optional payment-host allowlist —
/// the static form of the governed merchant registry.
///
/// `None` ⇒ no allowlist configured, so any (otherwise-valid) host is allowed
/// and only the SSRF deny-list guards the destination. `Some(list)` ⇒ **fail
/// closed**: the URL must parse and its lowercased host must be a member. An
/// unparseable URL or a URL with no host is rejected. The port is ignored — the
/// allowlist is keyed by host.
#[must_use]
pub fn host_allowed(url: &str, allowlist: Option<&[String]>) -> bool {
    let Some(list) = allowlist else {
        return true;
    };
    // reqwest re-exports the `url` crate; match the SSRF guard's parser.
    let Ok(parsed) = reqwest::Url::parse(url) else {
        return false;
    };
    // A URL with no host (or one that won't parse) is rejected — fail closed.
    parsed.host_str().is_some_and(|host| {
        let host = host.to_ascii_lowercase();
        list.iter().any(|allowed| allowed == &host)
    })
}

/// The pre-signing gate inputs to one proxied paid fetch: the SSRF policy, the
/// per-call ceiling, and the two pre-authorization layers.
pub struct Gates<'a> {
    /// SSRF policy for the destination check (production uses the default).
    pub ssrf_policy: SsrfPolicy,
    /// Per-call ceiling in the settlement token's base units, reserved against
    /// the conversation budget before signing; on settlement only the actual
    /// charged amount is committed and the remainder released.
    pub cap_base_units: u128,
    /// The granted pre-authorization ceiling: the per-call `cap_base_units`
    /// (and the session running total) must fit the per-transaction /
    /// per-session mandate, checked before the budget is reserved and before
    /// any credential is minted. [`SpendCap::unlimited`] disables it.
    pub spend_cap: &'a SpendCap,
    /// An optional, ALREADY-verified AP2-style mandate chain (see
    /// [`mandate::resolve`]): when present, the per-call spend, the destination
    /// host, and the approved args must all fit what the chain authorizes —
    /// checked at the same pre-sign seam as `spend_cap`. `None` (the default,
    /// and the only value when `TEMPO_MANDATE_ISSUER_PUBKEY` is unset) leaves
    /// behavior exactly as before.
    pub mandate: Option<&'a VerifiedMandateChain>,
}

/// One paid-fetch intent for [`fulfill_with`]: the two argument blobs the
/// approval binding compares, the destination, and the gate stack.
pub struct ProxyCall<'a> {
    /// The args the sandbox sent (opaque JSON; value-compared to the approved
    /// args — do not trust the sandbox to send what was approved).
    pub args_json: &'a str,
    /// The args the human approved (opaque JSON; the mandate args binding also
    /// checks against exactly these).
    pub approved_args_json: &'a str,
    /// The destination URL. Callers MUST derive this from the APPROVED args —
    /// never from a separately supplied (sandbox-controlled) value.
    pub url: &'a str,
    /// The HTTP method. Callers MUST derive this from the APPROVED args, same
    /// as `url`.
    pub method: reqwest::Method,
    /// Optional request body, sent verbatim. Callers MUST derive this from
    /// the APPROVED args, same as `url`.
    pub body: Option<&'a str>,
    /// Caller-supplied request headers. Callers MUST derive these from the
    /// APPROVED args, same as `url`, and must already have rejected any
    /// header the payment protocol itself owns (see
    /// `polyc_tools::paid_fetch::RESERVED_HEADER_NAMES`) — this layer does not
    /// re-check the name set.
    pub headers: &'a [(String, String)],
    /// The conversation whose aggregate budget the spend reserves against.
    pub conversation_id: &'a str,
    /// The reservation this call resolves at its settlement point.
    ///
    /// Its identity is minted by the caller from the approved call's own
    /// binding, so it is the same on a retry and the same on both durable write
    /// paths, and the commit that settles it is keyed by it. Its amount is the
    /// per-call ceiling being held. The caller keeps a clone and reads
    /// [`SharedReservation::settlement`] on the way out.
    pub reservation: SharedReservation,
    /// Durable cancellation net for the settlement record (#168b); `None`
    /// when no durable log is configured.
    pub settlement_sink: Option<Arc<dyn SettlementSink>>,
    /// The pre-signing gate stack.
    pub gates: Gates<'a>,
}

/// Fulfills a paid-fetch intent with an already-built [`PaymentProvider`]
/// (production passes a [`CappedProvider`] over the resolved signer; tests pass
/// a stub).
///
/// Applies the approval-binding, SSRF, and aggregate-budget gates, then performs
/// the 402-gated GET.
///
/// Never returns an error type — every outcome (success or machine-readable
/// rejection) is a [`ProxyOutcome`] so it maps cleanly onto the wire frame.
///
/// # The returned net is armed
///
/// The second member is the call's [`SettlementNet`], **armed**. The durable
/// record of a settled payment is the caller's own `.await`, so the net has to
/// span that write rather than end where this function does. Hold it across the
/// durable commit and the settlement receipt, and disarm it once both have
/// landed. Dropping it undisarmed writes the record itself, which the keyed
/// identity makes a replay rather than a second charge.
///
/// # Cancellation safety
///
/// Dropping this future is safe at every await point. The reservation is
/// resolved inside the provider decorator at the moment of settlement, rather
/// than on the way back out here, so a drop after a payment settled still leaves
/// the charge recorded — see [`SharedReservation`]. A drop before settlement
/// leaves the durable reservation open, which is the fail-closed direction and
/// what the authority's own deadline reclaims. The durable write is covered by
/// the net described above; see [`SettlementSink`].
pub async fn fulfill_with<P: PaymentProvider>(
    mut call: ProxyCall<'_>,
    provider: &P,
    spend: &dyn ConversationSpend,
) -> (ProxyOutcome, SettlementNet) {
    // Armed BEFORE the first `.await` and handed out still armed. Arming this
    // early is free — the net records nothing unless a payment really settled —
    // and it means no reachable drop point, here or in the caller, sits outside
    // it.
    let net = SettlementNet::arm(call.settlement_sink.take(), call.reservation.clone());
    let outcome = fulfill_call(call, provider, spend).await;
    // No `.await` between the call's return and this one, so a cancellation
    // cannot land where the net is neither inside the future nor with the caller.
    (outcome, net)
}

/// The paid-fetch exchange itself, under the net [`fulfill_with`] arms around
/// it.
///
/// `skip_all` + an explicit `conversation_id` field: the default `#[instrument]`
/// behavior records every argument via `Debug`, and `ProxyCall` carries the
/// destination URL and raw args JSON — not something a payments codepath
/// should risk logging wholesale. This gives every event in this function
/// (including the ones in callees like the SSRF/budget gates) the same
/// correlation id the explicit `warn!` below also stamps.
#[tracing::instrument(
    name = "payments.fulfill",
    level = "info",
    skip_all,
    fields(conversation_id = call.conversation_id)
)]
async fn fulfill_call<P: PaymentProvider>(
    call: ProxyCall<'_>,
    provider: &P,
    spend: &dyn ConversationSpend,
) -> ProxyOutcome {
    // (1) Approval binding: the args the sandbox sends must match the args the
    // human approved. Compare by VALUE (canonicalized JSON), not raw bytes — the
    // provider re-emits the call with non-deterministic key order, so a byte
    // compare rejects a legitimately-approved call (the same loop that bit the
    // HITL approval gate; see `polyc_crypto::canon`). Only ordering is
    // neutralized; different key/value sets still mismatch.
    if canon_args(call.args_json) != canon_args(call.approved_args_json) {
        return ProxyOutcome::rejected(RejectReason::ApprovalMismatch);
    }

    // (2) SSRF guard on the approved URL (async: DNS resolve runs off-worker).
    // Validate AND pin the approved address: the fetch below connects through a
    // resolver bound to exactly this IP, so the connection can't be rebound to an
    // internal host between check and connect. Redirects stay disabled.
    let resolver = match ssrf::pinned_resolver_async(call.url, call.gates.ssrf_policy).await {
        Ok(r) => r,
        Err(e) => return ProxyOutcome::rejected(RejectReason::BlockedDestination(e)),
    };
    let Ok(client) = pinned_http_client(resolver) else {
        return ProxyOutcome::rejected(RejectReason::ClientBuild);
    };

    // (3) Aggregate ceiling. Read the conversation's durable floor — committed
    // spend plus whatever other calls are holding — so the ceiling is enforced
    // across a control-plane restart and across replicas. An authority that
    // cannot answer refuses the payment: reporting zero would hand the
    // conversation its whole ceiling again.
    let Ok(floor) = spend.floor(call.conversation_id).await else {
        return ProxyOutcome::rejected(RejectReason::SpendUnreachable);
    };
    // (3) Pre-authorization (pre-signing spend cap): the intended per-call spend
    // must fit the granted per-transaction / per-session ceiling. The session
    // running total is the durable floor read just above, so spend from other
    // replicas and from before a restart counts toward the per-session mandate.
    // An over-cap request is rejected here — before anything is reserved and
    // before the signer runs — so no credential is ever minted for it.
    if let Err(e) = call
        .gates
        .spend_cap
        .authorize(call.gates.cap_base_units, floor)
    {
        return ProxyOutcome::rejected(RejectReason::OverSpendCap(e));
    }
    // (3) Pre-authorization (signed mandate chain): when a verified chain is
    // presented, the intended spend must also fit the chain's authorization —
    // amount, destination host, AND the exact approved args — before anything is
    // reserved and before the signer runs. `None` (mandates off or no chain
    // presented) changes nothing. The host binding uses the SAME parser the SSRF
    // guard and fetch use; a mandate presented for a URL with no parseable host
    // authorizes nothing (fail closed).
    if let Some(chain) = call.gates.mandate {
        let Some(host) = reqwest::Url::parse(call.url)
            .ok()
            .and_then(|u| u.host_str().map(str::to_owned))
        else {
            return ProxyOutcome::rejected(RejectReason::MandateHostUnknown);
        };
        if let Err(e) = chain.authorize(call.gates.cap_base_units, &host, call.approved_args_json) {
            return ProxyOutcome::rejected(RejectReason::MandateRefused(e));
        }
    }
    // Reserve the ceiling durably, before signing. A process that dies between
    // here and settlement leaves this reservation held, which over-counts an
    // in-flight charge rather than under-counting a real one; the authority's
    // own deadline reclaims it.
    let reservation = call.reservation;
    if let Err(refused) = spend
        .reserve(call.conversation_id, reservation.id(), reservation.amount())
        .await
    {
        return ProxyOutcome::rejected(match refused {
            ReserveRefused::Ceiling(exceeded) => RejectReason::OverBudget(exceeded),
            ReserveRefused::TooManyInFlight => RejectReason::TooManyPaymentsInFlight,
            ReserveRefused::AlreadyHeld => RejectReason::PaymentAlreadyInFlight,
            ReserveRefused::Unreachable(_) => RejectReason::SpendUnreachable,
        });
    }
    // The granted reservation is resolved by the SETTLEMENT POINT rather than by
    // this function on the way back out. That ordering is what makes the charge
    // cancellation-safe: a turn deadline can no longer fall between a settled
    // payment and the record of it, because the two are one uninterrupted step
    // inside `PaymentObserver::pay` (#168a).
    //
    // The DURABLE write (#168b) is covered by the `SettlementNet` `fulfill_with`
    // armed before it called this function and hands to the caller afterwards.
    // It is deliberately not armed and disarmed here: the write it stands in for
    // is the caller's, and a net that ended at this function's return would
    // cover everything except the await it exists for.

    // (4)+(5) Per-call cap + chain guard + onchain key limit are enforced by
    // the provider during the 402 exchange.
    let observed = PaymentObserver {
        inner: provider,
        reservation: reservation.clone(),
    };
    let request = OutboundRequest {
        method: call.method,
        url: call.url,
        body: call.body,
        headers: call.headers,
    };
    let result = paid_request_with_client(&observed, &request, &client).await;
    // A settled payment already resolved the reservation, inside `pay`. All that
    // is left is the free (non-402) case, where no `pay` ran and the headroom
    // must go back. Unconditional and a no-op after a settlement, so this
    // function never has to re-decide what settled.
    reservation.release();

    // Attribute the spend iff a payment was actually made — even if the upstream
    // returned no parseable receipt (a paid service that omits it must still
    // consume budget). A credential may already have been created (and the
    // payment may have settled) before a retry/body-read failure: the reservation
    // resolved at settlement, so the attribution rides the Rejected variant too —
    // hiding it would be a settle-but-fail bypass.
    //
    // The amount is the figure the reservation really recorded, so it is present
    // whenever a payment settled — including the case where the challenge amount
    // could not be read, which records the reserved ceiling. That closes the
    // known undercount the receipt-summing floor carried (#1739): a settled
    // charge can no longer reach the durable record without a number.
    let paid = match reservation.settlement() {
        Some(Settlement::Charged(amount)) => Some(PaidAttribution {
            amount_base_units: Some(amount),
            ..PaidAttribution::default()
        }),
        Some(Settlement::Unspent) | None => None,
    };
    match result {
        Ok(resp) => {
            let receipt = resp.receipt.as_ref().map(ProxyReceipt::from);
            ProxyOutcome::Fetched {
                status: resp.status,
                body: resp.body,
                receipt_header: resp.receipt_header,
                receipt,
                paid,
            }
        }
        Err(e) => {
            // The only place a `PaymentError`'s real detail is ever captured: the
            // reader-facing wording derived from it downstream (see
            // `polyc_control_plane::harness_dialer::payment_failure_text`) may stay
            // terse or redact specifics, but an operator must always be able to
            // recover the actual chain/provider reason from here.
            tracing::warn!(
                conversation_id = call.conversation_id,
                error = %e,
                "outbound payment attempt failed"
            );
            ProxyOutcome::Rejected {
                reason: RejectReason::Payment(e),
                paid,
            }
        }
    }
}

/// Body cap for a non-paying `web_fetch` (the harness applies a second,
/// turn-level cap on top).
const WEB_FETCH_BODY_CAP: usize = 100_000;

/// Per-call timeout for a non-paying `web_fetch`.
const WEB_FETCH_TIMEOUT_SECS: u64 = 20;

/// Fulfill a NON-paying `web_fetch` of `url`.
///
/// Applies the SAME SSRF guard (and optional host allowlist) as the paid path,
/// then GETs the URL on the trusted side and returns the status + a capped
/// body. The caller parses its tool's argument JSON and passes the URL.
///
/// No payment, no approval binding, no signing — `web_fetch` is read-only, so it
/// needs none of the wallet/budget machinery. Redirects are NOT followed (a
/// redirect-based SSRF bypass is blocked at the first hop, which the guard
/// already validated).
pub async fn fulfill_unpaid(
    url: &str,
    ssrf_policy: SsrfPolicy,
    allowlist: Option<&[String]>,
) -> ProxyOutcome {
    if !host_allowed(url, allowlist) {
        return ProxyOutcome::rejected(RejectReason::HostNotAllowed);
    }
    // Validate AND pin the approved address so the GET connects to exactly the
    // IP the guard classified — no connect-time DNS rebinding. Redirects are
    // disabled (a redirect-based bypass is blocked at the first, validated hop).
    let resolver = match ssrf::pinned_resolver_async(url, ssrf_policy).await {
        Ok(r) => r,
        Err(e) => return ProxyOutcome::rejected(RejectReason::BlockedDestination(e)),
    };
    // Route through the shared egress factory so the redirects-disabled SSRF
    // invariant lives in one place; the web_fetch path only differs by its
    // shorter request timeout.
    let Ok(client) = pinned_http_client_with_timeout(
        resolver,
        std::time::Duration::from_secs(WEB_FETCH_TIMEOUT_SECS),
    ) else {
        return ProxyOutcome::rejected(RejectReason::ClientBuild);
    };
    match client.get(url).send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            // The shared capped reader in TRUNCATE mode: stream chunks and stop
            // at the byte cap — never buffer the whole body, so a hostile/huge
            // upstream can't OOM the trusted control plane.
            polyc_egress::egress::read_capped_text(resp, WEB_FETCH_BODY_CAP, CapMode::Truncate)
                .await
                .map_or_else(
                    |_| ProxyOutcome::rejected(RejectReason::FetchFailed),
                    |body| ProxyOutcome::Fetched {
                        status,
                        body,
                        receipt_header: None,
                        receipt: None,
                        paid: None,
                    },
                )
        }
        Err(_) => ProxyOutcome::rejected(RejectReason::FetchFailed),
    }
}

/// One approved `paid_fetch` for the production [`fulfill`] entry point.
///
/// The caller (the control plane's harness dialer) parses the tool's argument
/// JSON itself and passes the derived `url`/`max_spend` — the tool schema stays
/// with the tool, not in this crate.
pub struct FulfillRequest<'a> {
    /// The args the sandbox sent (opaque JSON; value-compared to the approved
    /// args by the binding gate).
    pub args_json: &'a str,
    /// The args the human approved (opaque JSON).
    pub approved_args_json: &'a str,
    /// The destination URL, derived by the caller from the APPROVED args —
    /// never from a separately supplied (sandbox-controlled) value.
    pub url: &'a str,
    /// The HTTP method, derived by the caller from the APPROVED args (see
    /// `polyc_tools::paid_fetch::parse_args`). Defaults to `GET`.
    pub method: reqwest::Method,
    /// Optional request body, derived by the caller from the APPROVED args.
    /// Meaningful with `POST`.
    pub body: Option<&'a str>,
    /// Caller-supplied request headers, derived by the caller from the
    /// APPROVED args — already validated against
    /// `polyc_tools::paid_fetch::RESERVED_HEADER_NAMES` upstream of this
    /// struct.
    pub headers: &'a [(String, String)],
    /// The per-call spend ceiling (decimal dollar string) from the approved
    /// args, when the call carried one; falls back to the configured
    /// `TEMPO_MAX_SPEND`, then the built-in conservative default.
    pub max_spend: Option<&'a str>,
    /// Which signer pays: the caller's delegated wallet key or the
    /// deployment's env-configured signer.
    pub key_source: KeySource<'a>,
    /// The conversation whose aggregate budget the spend reserves against.
    pub conversation_id: &'a str,
    /// Stable identity of this call's reservation, minted by the caller from the
    /// approved call's own binding so it is the same on a retry and the same on
    /// both durable write paths. [`fulfill`] grants a
    /// [`SharedReservation`] under it at the per-call ceiling it resolves.
    pub reservation_id: &'a str,
    /// Current unix time (for keychain expiry checks and mandate expiry).
    pub now_unix: u64,
    /// An optional signed AP2-style Intent→Cart→Payment chain presented as
    /// pre-authorization. Engages ONLY when `TEMPO_MANDATE_ISSUER_PUBKEY` is
    /// configured (see [`mandate::resolve`]): unconfigured, any presented
    /// chain is ignored and behavior is exactly the HITL-approval + spend-cap
    /// path. Configured, a presented chain must verify end-to-end or the
    /// payment is refused before anything is signed.
    pub mandate_chain: Option<&'a MandateChain>,
    /// Durable pre-sign attempt recorder, run immediately before signing; a
    /// recorder failure fails the payment closed. `None` (dev/CI) records
    /// nothing.
    pub recorder: Option<polyc_payments_client::outbound::AttemptRecorder>,
    /// Durable cancellation net for the settlement receipt (#168b); `None`
    /// when no durable log is configured.
    pub settlement_sink: Option<Arc<dyn SettlementSink>>,
    /// The CALLER's own persona per-payment ceiling (`#514`/`#517`), a
    /// decimal dollar string, when their admin set one narrower than the
    /// deployment default. The most a SINGLE payment may cost — never a
    /// cumulative allowance over a window, and never read from (or written
    /// back to) the delegated key's own onchain, per-period allowance; the
    /// two are distinct quantities enforced by distinct authorities and are
    /// never converged into each other (#2365). `None` (no policy, or one
    /// with no limit set) changes nothing — only the deployment
    /// `max_spend`/`TEMPO_MAX_SPEND` cap applies, exactly as before. NARROWS
    /// the per-call cap (folded into the existing cap-min computation below
    /// `max_spend`); never widens it past the deployment cap.
    pub persona_per_payment_limit: Option<&'a str>,
    /// The CALLER's own persona host allowlist (`#514`/`#517`), when their
    /// admin set one. `None` or empty changes nothing — only the deployment
    /// `paid_host_allowlist` applies, exactly as before. NARROWS the
    /// destination check (the URL must pass both this AND the deployment
    /// allowlist); never widens it.
    pub persona_allowed_hosts: Option<&'a [String]>,
}

/// Resolves the enforced per-call spend ceiling, in base units: the TIGHTER of
/// an agent-authored `max_spend` and the operator's `TEMPO_MAX_SPEND`, never
/// the wider of the two. `TEMPO_MAX_SPEND` is a ceiling, not merely a default
/// an agent argument can override — when both are present the smaller wins;
/// when only one is present it alone applies; [`DEFAULT_CAP`] applies only
/// when NEITHER side supplied a value.
///
/// A value that WAS supplied but fails to parse (malformed, or a magnitude
/// overflowing `u128`) is rejected outright ([`RejectReason::InvalidMaxSpend`])
/// rather than silently falling back to a wider cap — the previous behavior
/// let a garbled agent-authored `max_spend` fall through to
/// [`DEFAULT_CAP`], which can be WIDER than an operator cap the agent's
/// malformed value was meant to fall under.
///
/// # Panics
///
/// Does not panic in practice: the only `expect` re-parses the compile-time
/// constant `DEFAULT_CAP` at `decimals`, and the config loader bounds decimals
/// to the tempo-MPP ceiling (`MAX_CURRENCY_DECIMALS`), so the conversion
/// always succeeds.
fn resolve_cap_base_units(
    req_max_spend: Option<&str>,
    cfg_max_spend: Option<&str>,
    decimals: u32,
) -> Result<u128, RejectReason> {
    let parse = |v: &str| {
        crate::amount::dollars_to_base_units(v, decimals).ok_or(RejectReason::InvalidMaxSpend)
    };
    let req_cap = req_max_spend.map(parse).transpose()?;
    let cfg_cap = cfg_max_spend.map(parse).transpose()?;
    Ok(match (req_cap, cfg_cap) {
        (Some(r), Some(c)) => r.min(c),
        (Some(r), None) => r,
        (None, Some(c)) => c,
        (None, None) => {
            crate::amount::dollars_to_base_units(DEFAULT_CAP, decimals).expect("default cap parses")
        }
    })
}

/// A refusal raised before the payment exchange began, with the inert net that
/// goes with it: nothing settled, so no net has anything to record.
const fn inert_rejection(reason: RejectReason) -> (ProxyOutcome, SettlementNet) {
    (ProxyOutcome::rejected(reason), SettlementNet::inert())
}

/// Production entry point for fulfilling an approved `paid_fetch`.
///
/// Resolves the outbound signer from `cfg` for the request's
/// [`KeySource`] (a per-persona delegated key or the deployment's env signer),
/// wraps it in a [`CappedProvider`] (per-call cap + chain guard), and fulfills
/// the intent via [`fulfill_with`].
///
/// # The returned net is armed
///
/// The second member is the call's [`SettlementNet`]. See [`fulfill_with`] for
/// what the caller owes it: hold it across the durable settlement writes, and
/// disarm it only once they have landed.
///
/// # Panics
///
/// Does not panic in practice — see the private `resolve_cap_base_units`
/// helper this function calls, just above.
pub async fn fulfill(
    req: FulfillRequest<'_>,
    cfg: &PaymentsConfig,
    spend: &dyn ConversationSpend,
) -> (ProxyOutcome, SettlementNet) {
    // Governed merchant allowlist (fail closed). Enforced on the APPROVED url and
    // before the signer is resolved — a destination off the allowlist never
    // reaches the signing path. `None` ⇒ no allowlist, only the SSRF guard.
    if !host_allowed(req.url, cfg.paid_host_allowlist.as_deref()) {
        return inert_rejection(RejectReason::HostNotAllowed);
    }
    // The caller's own persona host allowlist (`#514`/`#517`) NARROWS the
    // deployment allowlist above — it never widens it (the deployment check
    // already ran and passed). A distinct reject reason so the caller can word
    // "your admin restricted this wallet" apart from "the deployment restricted
    // outbound payments." `None` OR an EMPTY persona list both mean "no
    // per-persona restriction" (matching the `SpendPolicy.allowed_hosts` proto
    // contract) — unlike the deployment allowlist, where an empty `Some(&[])`
    // deliberately denies every host, an empty persona list is what an unset
    // (or explicitly cleared) policy looks like, so it must not fail closed.
    let persona_hosts = req.persona_allowed_hosts.filter(|h| !h.is_empty());
    if !host_allowed(req.url, persona_hosts) {
        return inert_rejection(RejectReason::PersonaHostNotAllowed);
    }
    // The reservation and the per-call cap are both in the settlement token's
    // base units (scaled at the configured decimals) — one unit, no lossy
    // conversion.
    let cap_base_units = match resolve_cap_base_units(
        req.max_spend,
        cfg.max_spend.as_deref(),
        cfg.currency_decimals,
    ) {
        Ok(units) => units,
        Err(reason) => return inert_rejection(reason),
    };
    // Fold the caller's own persona per-payment ceiling (`#514`/`#517`) into
    // the SAME cap the deployment default already produced — a pure
    // narrowing input to the existing reservation/`CappedProvider` machinery,
    // not a second enforcement path. A limit that fails to parse (never
    // happens for a value `wallet_nav` validated at set time) is ignored
    // rather than widening the cap. Never reads `period_secs` — this is a
    // flat per-payment ceiling, not a per-period one (#2365).
    let cap_base_units = req
        .persona_per_payment_limit
        .filter(|l| !l.is_empty())
        .and_then(|l| crate::amount::dollars_to_base_units(l, cfg.currency_decimals))
        .map_or(cap_base_units, |persona_cap| {
            persona_cap.min(cap_base_units)
        });
    // Granted pre-authorization (per-transaction / per-session). Parsed
    // and validated at config load, so this is infallible here; `unlimited` when
    // neither bound is configured.
    let spend_cap = cfg.presign_spend_cap();
    // Signed-mandate pre-authorization: verify the presented chain against the
    // configured issuer key, before the signer is resolved. `Ok(None)` (no
    // chain presented, or mandates unconfigured — the default) is today's
    // path; a presented-but-invalid chain refuses here, fail closed.
    let verified_mandate = match mandate::resolve(
        req.mandate_chain,
        cfg.mandate_issuer_public_key(),
        req.conversation_id,
        req.now_unix,
    ) {
        Ok(v) => v,
        Err(e) => return inert_rejection(RejectReason::MandateRefused(e)),
    };

    // A configured settlement currency + decimals are required on EVERY signing
    // path (keychain and gated raw/Direct): the per-call cap is denominated in
    // that token, so without it the cap can't be enforced — fail closed rather
    // than sign blind.
    let Some(currency) = cfg.currency.as_deref() else {
        return inert_rejection(RejectReason::MissingCurrency);
    };

    // Capture whose account pays before `key_source` is moved into resolution, so
    // a fulfilled payment reports the accurate payer rather than guessing.
    let payer_kind = req.key_source.payer_kind();
    let client = match cfg.resolve_client(req.key_source, req.now_unix).await {
        Ok(c) => c,
        Err(e) => return inert_rejection(RejectReason::BackendUnavailable(e)),
    };
    // The onchain account the resolved client settles from (the keychain wallet
    // or the raw signer's own address). Read before the client is wrapped in the
    // provider chain below.
    let paying_account = client.payer_address().map(str::to_owned);
    // Insert the durable pre-sign attempt recorder BELOW the cap (so it runs only
    // after cap/currency/chain validation, immediately before signing). A no-op
    // recorder is used when none is supplied (dev/CI), so the type is uniform.
    let recorder = req.recorder.unwrap_or_else(|| {
        std::sync::Arc::new(|_audit: polyc_payments_client::outbound::AttemptAudit| {
            Box::pin(async { Ok(()) }) as _
        })
    });
    // Innermost decorator: force self-settlement (a fully-signed `0x76` the
    // client pays its own gas for) when configured, so a fee-payer challenge
    // never produces a `0x78` envelope a facilitator has to co-sign. Sits below
    // the cap/recorder so those still see the merchant's original challenge.
    let self_settling = polyc_payments_client::outbound::ForceDirectSettle::new(
        client.provider().clone(),
        cfg.self_settle,
    );
    let signer_with_recorder =
        polyc_payments_client::outbound::AttemptRecordingProvider::new(self_settling, recorder);
    // Built from the already-narrowed `cap_base_units` (base units), not the
    // raw `cap` dollar string — the deployment cap folded with the caller's own
    // persona-policy limit above. Using `cap` here directly would silently
    // un-narrow the per-call signing cap back to the deployment default while
    // only the aggregate reservation stayed narrowed.
    let capped =
        match CappedProvider::from_base_units(signer_with_recorder, cap_base_units, currency) {
            Ok(c) => c.with_expected_chain_id(cfg.chain_id),
            Err(e) => return inert_rejection(RejectReason::Payment(e)),
        };

    let (mut outcome, net) = fulfill_with(
        ProxyCall {
            args_json: req.args_json,
            approved_args_json: req.approved_args_json,
            url: req.url,
            method: req.method,
            body: req.body,
            headers: req.headers,
            conversation_id: req.conversation_id,
            reservation: SharedReservation::granted(req.reservation_id, cap_base_units),
            settlement_sink: req.settlement_sink,
            gates: Gates {
                ssrf_policy: SsrfPolicy::default(),
                cap_base_units,
                spend_cap: &spend_cap,
                mandate: verified_mandate.as_ref(),
            },
        },
        &capped,
        spend,
    )
    .await;
    // Attribute a settled payment to the accurate payer: the onchain account,
    // its human-readable amount, and whose account it is. Present only when a
    // payment settled, so a free (non-402) fetch or a rejection carries no
    // misleading attribution.
    if let Some(paid) = outcome.paid_mut() {
        paid.paying_account = paying_account;
        paid.payer_kind = Some(payer_kind);
        paid.paid_amount = paid
            .amount_base_units
            .map(|a| crate::amount::format_settled_amount(a, cfg.currency_decimals));
    }
    // The net travels out with the outcome, still armed. No `.await` runs between
    // `fulfill_with`'s return and this one.
    (outcome, net)
}

#[cfg(test)]
mod tests {
    use super::*;
    use mpp::client::PaymentProvider;
    use mpp::{
        Base64UrlJson, MppError, PaymentChallenge, PaymentCredential, PaymentPayload,
        format_www_authenticate,
    };
    use wiremock::matchers::{header_exists, method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    use std::sync::Mutex;
    use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};

    use polyc_spend_policy::budget::SpendUnreachable;

    /// A conversation spend authority for the tests: one ceiling, and the
    /// headroom each conversation is holding under it.
    ///
    /// Stands in for the durable ledger the control plane supplies. It grants
    /// and refuses on the same terms — the ceiling counts what is already held —
    /// so a gate test drives the real refusal rather than a stub that always
    /// says yes.
    #[derive(Debug)]
    struct Ceiling {
        cap: u128,
        held: Mutex<std::collections::HashMap<String, u128>>,
        reachable: AtomicBool,
    }

    impl Ceiling {
        fn new(cap: u128) -> Self {
            Self {
                cap,
                held: Mutex::new(std::collections::HashMap::new()),
                reachable: AtomicBool::new(true),
            }
        }

        /// Base units this conversation is holding.
        fn held(&self, conversation_id: &str) -> u128 {
            *self.held.lock().unwrap().get(conversation_id).unwrap_or(&0)
        }

        /// Takes the authority offline, so every read and reservation refuses.
        fn go_dark(&self) {
            self.reachable.store(false, AtomicOrdering::SeqCst);
        }
    }

    #[async_trait::async_trait]
    impl ConversationSpend for Ceiling {
        async fn floor(&self, conversation_id: &str) -> Result<u128, SpendUnreachable> {
            if !self.reachable.load(AtomicOrdering::SeqCst) {
                return Err(SpendUnreachable);
            }
            Ok(self.held(conversation_id))
        }

        async fn reserve(
            &self,
            conversation_id: &str,
            _reservation_id: &str,
            amount: u128,
        ) -> Result<(), ReserveRefused> {
            if !self.reachable.load(AtomicOrdering::SeqCst) {
                return Err(ReserveRefused::Unreachable(SpendUnreachable));
            }
            let mut held = self.held.lock().unwrap();
            let already = *held.get(conversation_id).unwrap_or(&0);
            let next = already.saturating_add(amount);
            let outcome = if next > self.cap {
                Err(ReserveRefused::Ceiling(
                    polyc_spend_policy::budget::BudgetExceeded {
                        requested: amount,
                        already,
                        cap: self.cap,
                    },
                ))
            } else {
                held.insert(conversation_id.to_owned(), next);
                Ok(())
            };
            drop(held);
            outcome
        }
    }

    #[test]
    fn host_allowed_open_when_no_allowlist() {
        // No allowlist configured ⇒ any well-formed URL passes (SSRF still guards).
        assert!(host_allowed("https://api.example.com/x", None));
        assert!(host_allowed("https://anything.io", None));
    }

    #[test]
    fn host_allowed_member_passes_case_and_port_insensitive() {
        let list = vec!["api.example.com".to_string(), "data.bar.io".to_string()];
        assert!(host_allowed(
            "https://api.example.com/path?q=1",
            Some(&list)
        ));
        // Case-insensitive host match.
        assert!(host_allowed("https://API.Example.COM/x", Some(&list)));
        // Port is ignored — the allowlist is keyed by host.
        assert!(host_allowed("https://data.bar.io:8443/y", Some(&list)));
    }

    #[test]
    fn host_allowed_non_member_rejected() {
        let list = vec!["api.example.com".to_string()];
        assert!(!host_allowed("https://evil.example.com/x", Some(&list)));
        // A subdomain is not an implicit member — exact host only.
        assert!(!host_allowed("https://sub.api.example.com/x", Some(&list)));
    }

    #[test]
    fn host_allowed_fails_closed_on_bad_url() {
        let list = vec!["api.example.com".to_string()];
        // Unparseable URL or no host ⇒ rejected (fail closed) when an allowlist
        // is set.
        assert!(!host_allowed("not a url", Some(&list)));
        assert!(!host_allowed("", Some(&list)));
    }

    #[test]
    fn host_allowed_empty_list_denies_all() {
        // A configured-but-empty allowlist (a set-but-degenerate
        // TEMPO_PAID_HOST_ALLOWLIST) denies every host — the control fails
        // closed instead of silently turning off.
        assert!(!host_allowed("https://api.example.com/x", Some(&[])));
    }

    #[test]
    fn host_allowed_rejects_encoding_and_control_tricks() {
        // The bypass classes shipped egress relays have had (#597): control
        // bytes and percent-encoding in the destination must never match an
        // allowlisted host — and a parse failure is a refusal, not a pass.
        let list = ["api.example.com".to_owned()];
        assert!(!host_allowed(
            "https://api.example.com%00.evil.test/x",
            Some(&list)
        ));
        assert!(!host_allowed(
            "https://api.example.com\u{0}.evil.test/x",
            Some(&list)
        ));
        assert!(!host_allowed(
            "https://api%2eexample%2ecom.evil.test/x",
            Some(&list)
        ));
        // A lookalike suffix is a different host.
        assert!(!host_allowed(
            "https://api.example.com.evil.test/x",
            Some(&list)
        ));
        // Schemes with no host fail closed.
        assert!(!host_allowed("data:text/plain,hi", Some(&list)));
        assert!(!host_allowed("file:///etc/passwd", Some(&list)));
    }

    #[test]
    fn host_allowed_userinfo_cannot_spoof_host() {
        // The host is what the URL parser resolves, not what precedes the `@`:
        // an allowlisted name in the userinfo position must not admit a
        // different real host (the same parser feeds the SSRF guard and fetch).
        let list = vec!["api.example.com".to_string()];
        assert!(!host_allowed(
            "https://api.example.com@evil.io/x",
            Some(&list)
        ));
        assert!(host_allowed("https://user@api.example.com/x", Some(&list)));
    }

    /// A provider that never satisfies a challenge — used so the proxy's gate
    /// logic is exercised without any chain/signing. For a 200 response the
    /// provider is never asked to pay.
    #[derive(Clone)]
    struct NoopProvider;

    impl PaymentProvider for NoopProvider {
        fn supports(&self, _method: &str, _intent: &str) -> bool {
            false
        }
        async fn pay(&self, _challenge: &PaymentChallenge) -> Result<PaymentCredential, MppError> {
            Err(MppError::InvalidConfig("noop provider cannot pay".into()))
        }
    }

    /// A provider that satisfies the tempo/charge challenge with a canned
    /// credential, so a 402→pay→retry round-trips without a chain call.
    #[derive(Clone)]
    struct PayingStub;

    impl PaymentProvider for PayingStub {
        fn supports(&self, _method: &str, _intent: &str) -> bool {
            true
        }
        async fn pay(&self, challenge: &PaymentChallenge) -> Result<PaymentCredential, MppError> {
            Ok(PaymentCredential::new(
                challenge.to_echo(),
                PaymentPayload::hash("0xstub"),
            ))
        }
    }

    /// A provider whose `pay` never returns: it parks forever mid-settlement, so
    /// a test can cancel the calling future at the one await point where a
    /// credential is being minted but does not yet exist.
    #[derive(Clone)]
    struct HangingPayStub;

    impl PaymentProvider for HangingPayStub {
        fn supports(&self, _method: &str, _intent: &str) -> bool {
            true
        }
        async fn pay(&self, _challenge: &PaymentChallenge) -> Result<PaymentCredential, MppError> {
            std::future::pending().await
        }
    }

    /// A provider that accepts the challenge (so `pay` is actually invoked) but
    /// always fails to produce a credential — drives `paid_get_with_client` into
    /// `HttpError::Payment`, which `map_http_error` turns into
    /// [`PaymentError::Verification`]. Used to exercise the failure-logging path,
    /// distinct from [`NoopProvider`] (which never even attempts `pay`).
    #[derive(Clone)]
    struct FailingPayProvider;

    impl PaymentProvider for FailingPayProvider {
        fn supports(&self, _method: &str, _intent: &str) -> bool {
            true
        }
        async fn pay(&self, _challenge: &PaymentChallenge) -> Result<PaymentCredential, MppError> {
            Err(MppError::InvalidConfig(
                "stubbed chain failure: key rejected".into(),
            ))
        }
    }

    /// A paying stub that counts the credentials it minted, so a test can tell
    /// "the provider never signed" apart from "the provider signed and the proxy
    /// refused to hand the credential over."
    #[derive(Clone, Default)]
    struct CountingPayStub(Arc<std::sync::atomic::AtomicUsize>);

    impl CountingPayStub {
        fn minted(&self) -> usize {
            self.0.load(AtomicOrdering::SeqCst)
        }
    }

    impl PaymentProvider for CountingPayStub {
        fn supports(&self, _method: &str, _intent: &str) -> bool {
            true
        }
        async fn pay(&self, challenge: &PaymentChallenge) -> Result<PaymentCredential, MppError> {
            self.0.fetch_add(1, AtomicOrdering::SeqCst);
            Ok(PaymentCredential::new(
                challenge.to_echo(),
                PaymentPayload::hash("0xstub"),
            ))
        }
    }

    /// A `WWW-Authenticate: Payment …` header for a tempo/charge challenge with
    /// an explicit id and base-unit amount, so a test can drive two DISTINCT
    /// challenges through one call (the 402 exchange dedupes by challenge id).
    fn challenge_header_for(id: &str, amount_base_units: &str) -> String {
        let request = Base64UrlJson::from_value(
            &serde_json::json!({ "amount": amount_base_units, "currency": PATH_USD }),
        )
        .unwrap();
        let challenge = PaymentChallenge::new(id, "polychrome", "tempo", "charge", request);
        format_www_authenticate(&challenge).unwrap()
    }

    /// A `WWW-Authenticate: Payment …` header for a tempo/charge challenge.
    fn challenge_header() -> String {
        // A real base-unit charge ($0.01 = 10000 at 6 decimals) carrying pathUSD,
        // so the observer captures the charged amount.
        let request = Base64UrlJson::from_value(
            &serde_json::json!({ "amount": "10000", "currency": PATH_USD }),
        )
        .unwrap();
        let challenge = PaymentChallenge::new("proxy-c1", "polychrome", "tempo", "charge", request);
        format_www_authenticate(&challenge).unwrap()
    }

    /// Mounts a 402-then-200 endpoint at `/paid`; the 200 includes a
    /// Payment-Receipt header only when `with_receipt`.
    async fn mount_402_then_200(server: &MockServer, with_receipt: bool) {
        let mut ok = ResponseTemplate::new(200).set_body_string("paid-body");
        if with_receipt {
            let header = mpp::format_receipt(&mpp::Receipt::success("tempo", "0xabc123")).unwrap();
            ok = ok.insert_header(mpp::PAYMENT_RECEIPT_HEADER, header.as_str());
        }
        Mock::given(method("GET"))
            .and(path("/paid"))
            .and(header_exists("authorization"))
            .respond_with(ok)
            .with_priority(1)
            .mount(server)
            .await;
        Mock::given(method("GET"))
            .and(path("/paid"))
            .respond_with(
                ResponseTemplate::new(402)
                    .insert_header("www-authenticate", challenge_header().as_str()),
            )
            .with_priority(5)
            .mount(server)
            .await;
    }

    fn args(url: &str) -> String {
        format!(r#"{{"url":"{url}"}}"#)
    }

    /// Mounts a 402-then-200 endpoint at `/paid` that only matches a POST
    /// carrying `expected_body` verbatim and the `x-custom` header — proves
    /// the SAME body/header set survives BOTH the pre-pay 402 probe and the
    /// post-pay retry (`AttemptRecordingProvider`/`CappedProvider` re-drive
    /// the request after signing), not just one of the two.
    async fn mount_402_then_200_post(server: &MockServer, expected_body: &str) {
        use wiremock::matchers::{body_string, header};
        let ok = ResponseTemplate::new(200).set_body_string("paid-body");
        Mock::given(method("POST"))
            .and(path("/paid"))
            .and(body_string(expected_body))
            .and(header("x-custom", "1"))
            .and(header_exists("authorization"))
            .respond_with(ok)
            .with_priority(1)
            .mount(server)
            .await;
        Mock::given(method("POST"))
            .and(path("/paid"))
            .and(body_string(expected_body))
            .and(header("x-custom", "1"))
            .respond_with(
                ResponseTemplate::new(402)
                    .insert_header("www-authenticate", challenge_header().as_str()),
            )
            .with_priority(5)
            .mount(server)
            .await;
    }

    /// A [`ProxyCall`] with the permissive test SSRF policy, no mandate, and no
    /// settlement sink — the common shape most gate tests start from.
    fn basic_call<'a>(
        args_json: &'a str,
        approved_args_json: &'a str,
        url: &'a str,
        conversation_id: &'a str,
        cap_base_units: u128,
        spend_cap: &'a SpendCap,
    ) -> ProxyCall<'a> {
        ProxyCall {
            args_json,
            approved_args_json,
            url,
            method: reqwest::Method::GET,
            body: None,
            headers: &[],
            conversation_id,
            reservation: SharedReservation::granted("res-1", cap_base_units),
            settlement_sink: None,
            gates: Gates {
                ssrf_policy: SsrfPolicy::permissive_for_tests(),
                cap_base_units,
                spend_cap,
                mandate: None,
            },
        }
    }

    /// The charged base-unit amount from a settled outcome, if any.
    fn charged(out: &ProxyOutcome) -> Option<u128> {
        out.paid().and_then(|p| p.amount_base_units)
    }

    #[tokio::test]
    async fn rejects_when_args_do_not_match_approval() {
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let sandbox_args = args("https://example.com/a");
        let approved_args = args("https://example.com/DIFFERENT");
        let (out, _net) = fulfill_with(
            basic_call(
                &sandbox_args,
                &approved_args,
                "https://example.com/DIFFERENT",
                "conv1",
                100_000,
                &cap,
            ),
            &NoopProvider,
            &spend,
        )
        .await;
        assert!(!out.is_ok());
        assert!(matches!(out.reason(), Some(RejectReason::ApprovalMismatch)));
        assert_eq!(
            spend.held("conv1"),
            0,
            "no reservation on approval mismatch"
        );
    }

    /// A non-URL argument is covered by the approval binding.
    ///
    /// `rejects_when_args_do_not_match_approval` varies the URL, so it cannot
    /// show that the comparison reaches any other field. `canon_args`
    /// canonicalizes both sides and compares the whole value set, so it should
    /// — this pins that, using `method` as the field that differs.
    ///
    /// Scope, stated honestly: this does NOT prove the effective HTTP method
    /// cannot be swapped. `basic_call` fixes the call's method, and production
    /// parses the method from the APPROVED args
    /// (`crates/control-plane/src/harness_dialer.rs`, `parse_args` over
    /// `approved.args_json`), so a sandbox-supplied method never becomes the
    /// effective one even with this comparison deleted. The read-cannot-become-
    /// a-write property lives there and is not tested here.
    #[tokio::test]
    async fn rejects_when_the_method_does_not_match_approval() {
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let approved_args = r#"{"url":"https://example.com/a","method":"GET"}"#;
        let sandbox_args = r#"{"url":"https://example.com/a","method":"POST"}"#;
        let (out, _net) = fulfill_with(
            basic_call(
                sandbox_args,
                approved_args,
                "https://example.com/a",
                "conv1",
                100_000,
                &cap,
            ),
            &NoopProvider,
            &spend,
        )
        .await;
        assert!(!out.is_ok());
        assert!(
            matches!(out.reason(), Some(RejectReason::ApprovalMismatch)),
            "a method the approver never saw must not execute"
        );
        assert_eq!(
            spend.held("conv1"),
            0,
            "no reservation on approval mismatch"
        );
    }

    /// The control for [`rejects_when_the_method_does_not_match_approval`]:
    /// the same args carrying the same `method` are NOT refused. Without this,
    /// that case would also pass if the gate rejected every call naming a
    /// method at all.
    ///
    /// The URL points at a local mock server, not a public host. A matching
    /// binding falls through to a real fetch, and a bare public URL would make
    /// this unit test egress to the live internet on every run — and worse,
    /// would let it PASS on an offline runner, where the fetch fails as a
    /// blocked destination and the absence of `ApprovalMismatch` proves
    /// nothing. Asserting the call actually succeeds closes that hole.
    #[tokio::test]
    async fn accepts_a_method_the_approver_did_see() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;
        let url = format!("{}/a", server.uri());
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let same = format!(r#"{{"url":"{url}","method":"GET"}}"#);
        let (out, _net) = fulfill_with(
            basic_call(&same, &same, &url, "conv-method-control", 100_000, &cap),
            &NoopProvider,
            &spend,
        )
        .await;
        assert!(
            out.is_ok(),
            "matching args must clear the gate and reach the destination, got: {:?}",
            out.reason()
        );
    }

    #[tokio::test]
    async fn accepts_args_with_reordered_keys() {
        // Same args, DIFFERENT JSON key order → the binding must accept (compare
        // by value via `canon_args`), not reject. A raw byte compare here is the
        // same loop that bit the HITL service_create gate. The binding passing is
        // proven by the absence of the ApprovalMismatch rejection.
        //
        // The URL points at a local mock server, not a public host: a matching
        // binding falls through to a real fetch, and a bare public URL would make
        // this unit test egress to the live internet on every run.
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;
        let url = format!("{}/a", server.uri());
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let sandbox_args = format!(r#"{{"url":"{url}","max_spend":"$0.01"}}"#);
        let approved_args = format!(r#"{{"max_spend":"$0.01","url":"{url}"}}"#);
        let (out, _net) = fulfill_with(
            basic_call(
                &sandbox_args,
                &approved_args,
                &url,
                "conv-reorder",
                100_000,
                &cap,
            ),
            &NoopProvider,
            &spend,
        )
        .await;
        assert!(
            !matches!(out.reason(), Some(RejectReason::ApprovalMismatch)),
            "reordered-but-equal args must pass the approval binding, got: {:?}",
            out.reason()
        );
    }

    #[tokio::test]
    async fn rejects_ssrf_destination() {
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let a = args("https://169.254.169.254/latest/meta-data/");
        let mut call = basic_call(
            &a,
            &a,
            "https://169.254.169.254/latest/meta-data/",
            "conv1",
            100_000,
            &cap,
        );
        call.gates.ssrf_policy = SsrfPolicy::default();
        let (out, _net) = fulfill_with(call, &NoopProvider, &spend).await;
        assert!(!out.is_ok());
        assert!(matches!(
            out.reason(),
            Some(RejectReason::BlockedDestination(_))
        ));
        assert_eq!(spend.held("conv1"), 0, "no reservation when SSRF-blocked");
    }

    #[tokio::test]
    async fn unpaid_fetches_without_payment() {
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200).set_body_string("hello web"))
            .mount(&server)
            .await;
        let url = format!("{}/page", server.uri());
        let out = fulfill_unpaid(&url, SsrfPolicy::permissive_for_tests(), None).await;
        assert!(
            out.is_ok(),
            "unpaid fetch should succeed: {:?}",
            out.reason()
        );
        assert_eq!(out.status(), Some(200));
        assert_eq!(out.body(), Some("hello web"));
        assert!(out.paid().is_none(), "web_fetch never pays");
    }

    #[tokio::test]
    async fn unpaid_does_not_follow_redirects() {
        // The web_fetch client (built via the shared egress factory) keeps
        // automatic redirects disabled: a 302 is a classic SSRF bypass — validate
        // a benign host, then bounce to an internal one. The proxy must surface
        // the 302 itself, never chase its `Location`.
        let server = MockServer::start().await;
        let location = format!("{}/internal", server.uri());
        Mock::given(method("GET"))
            .and(path("/start"))
            .respond_with(ResponseTemplate::new(302).insert_header("location", location.as_str()))
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/internal"))
            .respond_with(ResponseTemplate::new(200).set_body_string("FOLLOWED"))
            .mount(&server)
            .await;
        let url = format!("{}/start", server.uri());
        let out = fulfill_unpaid(&url, SsrfPolicy::permissive_for_tests(), None).await;
        assert!(
            out.is_ok(),
            "fetch should return the redirect response itself: {:?}",
            out.reason()
        );
        assert_eq!(out.status(), Some(302), "redirect must NOT be followed");
        assert_ne!(out.body(), Some("FOLLOWED"));
    }

    #[tokio::test]
    async fn unpaid_rejects_ssrf_destination() {
        let out = fulfill_unpaid(
            "https://169.254.169.254/latest/meta-data/",
            SsrfPolicy::default(),
            None,
        )
        .await;
        assert!(!out.is_ok());
        assert!(matches!(
            out.reason(),
            Some(RejectReason::BlockedDestination(_))
        ));
    }

    #[tokio::test]
    async fn unpaid_rejects_off_allowlist_host() {
        let out = fulfill_unpaid(
            "https://evil.example.com/x",
            SsrfPolicy::permissive_for_tests(),
            Some(&["good.example.com".to_owned()]),
        )
        .await;
        assert!(!out.is_ok());
        assert!(matches!(out.reason(), Some(RejectReason::HostNotAllowed)));
    }

    #[tokio::test]
    async fn rejects_over_budget() {
        let spend = Ceiling::new(50_000); // 0.05
        let cap = SpendCap::unlimited();
        // Loopback, not a public host: the reservation is rejected before the
        // fetch, but the SSRF guard still resolves the URL first — a public host
        // would fire a live DNS lookup from this unit test.
        let a = args("http://127.0.0.1:9/x");
        let (out, _net) = fulfill_with(
            basic_call(
                &a,
                &a,
                "http://127.0.0.1:9/x",
                "conv1",
                100_000, // 0.10 > 0.05 cap
                &cap,
            ),
            &NoopProvider,
            &spend,
        )
        .await;
        assert!(!out.is_ok());
        assert!(matches!(out.reason(), Some(RejectReason::OverBudget(_))));
        assert_eq!(spend.held("conv1"), 0, "rejected reservation not counted");
    }

    #[tokio::test]
    async fn free_fetch_releases_budget() {
        // A 200 with no Payment-Receipt = a free (non-402) fetch: no payment
        // settled, so the reservation must be RELEASED — a free fetch never
        // consumes payment budget.
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/data"))
            .respond_with(ResponseTemplate::new(200).set_body_string("hello"))
            .mount(&server)
            .await;
        let url = format!("{}/data", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        let held = call.reservation.clone();
        let (out, _net) = fulfill_with(call, &NoopProvider, &spend).await;
        assert!(out.is_ok(), "expected success, got {:?}", out.reason());
        assert_eq!(out.status(), Some(200));
        assert_eq!(out.body(), Some("hello"));
        assert!(out.paid().is_none(), "a free fetch made no payment");
        assert_eq!(
            held.settlement(),
            Some(Settlement::Unspent),
            "a free fetch must resolve its reservation as unspent"
        );
        // The durable reservation is still held: `fulfill_with` resolves the
        // reservation, and the CALLER returns the headroom durably from that
        // answer. Splitting the two is what keeps the resolution synchronous at
        // the settlement point (#168a).
        assert_eq!(
            spend.held("conv1"),
            100_000,
            "the durable reservation is the caller's to return"
        );
    }

    #[tokio::test]
    async fn paid_fetch_commits_budget() {
        // A real 402→pay→200 (provider's pay() invoked) commits the reservation.
        let server = MockServer::start().await;
        mount_402_then_200(&server, true).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", 100_000, &cap),
            &PayingStub,
            &spend,
        )
        .await;
        assert!(out.is_ok(), "expected success, got {:?}", out.reason());
        assert!(out.receipt().is_some(), "a receipt must be parsed");
        assert_eq!(
            charged(&out),
            Some(10_000),
            "the charged base-unit amount is captured from the challenge"
        );
        assert_eq!(
            spend.held("conv1"),
            100_000,
            "the reservation holds the ceiling until the caller commits the charge"
        );
    }

    // #149: `paid_fetch` gains POST/body/header support — the SAME body and
    // caller header must survive both the initial 402 probe and the post-pay
    // retry (`mount_402_then_200_post` asserts on both), proving the request
    // spec is reused across the whole 402-challenge/sign/retry cycle, not
    // just the first leg.
    #[tokio::test]
    async fn paid_fetch_post_preserves_body_and_headers_through_the_retry() {
        let server = MockServer::start().await;
        let body = r#"{"a":1}"#;
        mount_402_then_200_post(&server, body).await;
        let url = format!("{}/paid", server.uri());
        let a = format!(r#"{{"url":"{url}","method":"POST","body":{body:?}}}"#);
        let headers = vec![("x-custom".to_owned(), "1".to_owned())];

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            ProxyCall {
                args_json: &a,
                approved_args_json: &a,
                url: &url,
                method: reqwest::Method::POST,
                body: Some(body),
                headers: &headers,
                conversation_id: "conv1",
                reservation: SharedReservation::granted("res-1", 100_000),
                settlement_sink: None,
                gates: Gates {
                    ssrf_policy: SsrfPolicy::permissive_for_tests(),
                    cap_base_units: 100_000,
                    spend_cap: &cap,
                    mandate: None,
                },
            },
            &PayingStub,
            &spend,
        )
        .await;
        assert!(
            out.is_ok(),
            "expected success (proves the mock matched on BOTH the 402 probe \
             and the post-pay retry), got {:?}",
            out.reason()
        );
        assert_eq!(
            charged(&out),
            Some(10_000),
            "the charged base-unit amount is captured from the challenge"
        );
    }

    #[tokio::test]
    async fn paid_fetch_without_receipt_still_commits_budget() {
        // Bypass guard: a service that CHARGES (402→pay→200) but returns NO
        // Payment-Receipt must STILL consume budget — commit keys on pay() being
        // invoked, not on receipt presence.
        let server = MockServer::start().await;
        mount_402_then_200(&server, false).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", 100_000, &cap),
            &PayingStub,
            &spend,
        )
        .await;
        assert!(out.is_ok(), "expected success, got {:?}", out.reason());
        assert!(out.receipt().is_none(), "no receipt header was returned");
        assert!(
            out.paid().is_some(),
            "a payment was made (pay invoked) — must carry attribution for audit"
        );
        assert_eq!(
            charged(&out),
            Some(10_000),
            "a payment with no receipt must still record the charged amount (no bypass)"
        );
    }

    #[tokio::test]
    async fn paid_then_retry_failure_still_commits_budget() {
        // 402 → pay (credential created) → the authed retry returns an over-cap
        // body, so the body read FAILS. A credential exists (payment may have
        // settled), so budget must COMMIT and the attribution must ride the
        // Rejected outcome (durable spend record) — releasing here would be a
        // settle-but-fail bypass.
        let server = MockServer::start().await;
        let oversize = vec![b'x'; 9 * 1024 * 1024]; // > the 8 MiB body cap
        Mock::given(method("GET"))
            .and(path("/paid"))
            .and(header_exists("authorization"))
            .respond_with(ResponseTemplate::new(200).set_body_bytes(oversize))
            .with_priority(1)
            .mount(&server)
            .await;
        Mock::given(method("GET"))
            .and(path("/paid"))
            .respond_with(
                ResponseTemplate::new(402)
                    .insert_header("www-authenticate", challenge_header().as_str()),
            )
            .with_priority(5)
            .mount(&server)
            .await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", 100_000, &cap),
            &PayingStub,
            &spend,
        )
        .await;
        assert!(!out.is_ok(), "the over-cap body read fails → error result");
        assert!(matches!(
            out.reason(),
            Some(RejectReason::Payment(PaymentError::BodyTooLarge { .. }))
        ));
        assert!(
            out.paid().is_some(),
            "a credential was created → must carry attribution for accounting"
        );
        assert_eq!(
            charged(&out),
            Some(10_000),
            "settle-but-fail must NOT release the charge; it records the charged amount (no bypass)"
        );
    }

    #[tracing_test::traced_test]
    #[tokio::test]
    async fn verification_failure_logs_the_real_reason() {
        // A provider that fails to produce a credential drives the proxy into
        // `PaymentError::Verification`, wrapping the real chain/provider reason.
        // `fulfill_with` must log that reason (crates/payments/src/proxy.rs) —
        // the ONLY place it's ever recoverable, since the reader-facing wording
        // derived from it downstream may stay terse. `#[traced_test]` (not a
        // hand-rolled `tracing::subscriber::set_default`) because plain
        // thread-scoped defaults race tracing's process-global per-callsite
        // `Interest` cache under `cargo test`'s default parallel execution.
        let server = MockServer::start().await;
        mount_402_then_200(&server, false).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv-verify", 100_000, &cap),
            &FailingPayProvider,
            &spend,
        )
        .await;
        assert!(matches!(
            out.reason(),
            Some(RejectReason::Payment(PaymentError::Verification(_)))
        ));

        assert!(logs_contain("outbound payment attempt failed"));
        assert!(
            logs_contain("conv-verify"),
            "expected the conversation id for correlation"
        );
        assert!(
            logs_contain("stubbed chain failure: key rejected"),
            "expected the REAL underlying reason, not a discarded/generic one"
        );
    }

    /// Mounts a `/paid` endpoint whose 402 challenge is answerable but whose
    /// authed retry never returns, so the calling future can be cancelled with a
    /// credential already minted and in flight to the merchant.
    async fn mount_402_then_hang(server: &MockServer) {
        Mock::given(method("GET"))
            .and(path("/paid"))
            .and(header_exists("authorization"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_string("late")
                    .set_delay(std::time::Duration::from_secs(30)),
            )
            .with_priority(1)
            .mount(server)
            .await;
        Mock::given(method("GET"))
            .and(path("/paid"))
            .respond_with(
                ResponseTemplate::new(402)
                    .insert_header("www-authenticate", challenge_header().as_str()),
            )
            .with_priority(5)
            .mount(server)
            .await;
    }

    #[tokio::test]
    async fn budget_is_charged_at_settlement_not_at_return() {
        // The design claim, stated directly: the charge lands on the budget the
        // moment the payment settles, NOT when `fulfill_with` returns. Here the
        // future is still alive and suspended in the hung authed retry — it has
        // not been cancelled, and has not returned — and the spend is already
        // counted. That is what removes the cancellation window rather than
        // cleaning up after it: there is no interval for a drop to fall into.
        let server = MockServer::start().await;
        mount_402_then_hang(&server).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        let held = call.reservation.clone();
        let mut fut = Box::pin(fulfill_with(call, &PayingStub, &spend));
        // Drive the future only as far as the hung retry, then stop polling it
        // WITHOUT dropping it.
        let polled = tokio::time::timeout(std::time::Duration::from_millis(300), &mut fut).await;
        assert!(
            polled.is_err(),
            "the hung retry must keep the future suspended past settlement"
        );
        assert_eq!(
            held.settlement(),
            Some(Settlement::Charged(10_000)),
            "the charge is recorded while the call is still in flight"
        );
        // And cancelling from here changes nothing, because nothing was left to do.
        drop(fut);
        assert_eq!(
            held.settlement(),
            Some(Settlement::Charged(10_000)),
            "cancelling after settlement cannot unrecord money already spent"
        );
    }

    #[tokio::test]
    async fn cancelled_during_settlement_leaves_no_phantom_charge() {
        // The other half of the property. Cancel INSIDE `pay`, while the
        // credential is being minted and does not yet exist: nothing can have
        // reached the merchant, so the reservation must go back in full. A
        // mechanism that committed on any in-flight payment would strand the
        // conversation's budget on a charge that never happened.
        let server = MockServer::start().await;
        mount_402_then_hang(&server).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        let held = call.reservation.clone();
        let fut = fulfill_with(call, &HangingPayStub, &spend);
        let out = tokio::time::timeout(std::time::Duration::from_millis(300), fut).await;
        assert!(
            out.is_err(),
            "the hung `pay` must outlast the deadline so the future is dropped mid-settlement"
        );
        assert_eq!(
            held.settlement(),
            None,
            "a drop BEFORE a credential exists must record no charge at all"
        );
        // The durable reservation stays held, because nothing in a dropped future
        // can release it — a `Drop` cannot await, and a release that only ran in
        // the surviving process would refund a charge that may have happened.
        // Over-counting an in-flight charge is the fail-closed direction, and the
        // authority's own deadline is what reclaims it.
        assert_eq!(
            spend.held("conv1"),
            100_000,
            "the durable reservation must stay held for the reaper"
        );
    }

    #[tokio::test]
    async fn cancelled_after_settlement_commits_budget() {
        // #168(a) end to end: 402 → pay (credential created) → the authed retry
        // HANGS. The turn future is dropped (deadline) mid-retry, so control never
        // returns to `fulfill_with` — yet the charged spend must stay counted
        // rather than rebound. Under a stack-RAII commit at the return this would
        // go back to 0 and re-approved calls would walk past the conversation
        // budget.
        let server = MockServer::start().await;
        mount_402_then_hang(&server).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        let held = call.reservation.clone();
        let fut = fulfill_with(call, &PayingStub, &spend);
        // Cancel the future while the authed retry hangs (a turn-deadline drop
        // between settlement and the durable write).
        let out = tokio::time::timeout(std::time::Duration::from_millis(300), fut).await;
        assert!(
            out.is_err(),
            "the hung retry must outlast the deadline so the future is dropped mid-settlement"
        );
        assert_eq!(
            held.settlement(),
            Some(Settlement::Charged(10_000)),
            "a deadline drop AFTER settlement must RECORD the charged spend, not release it"
        );
    }

    /// A [`SettlementSink`] stub that records each persisted amount, so a test can
    /// assert the cancellation net fired with the charged spend.
    #[derive(Clone, Default)]
    struct RecordingSink(std::sync::Arc<std::sync::Mutex<Vec<u128>>>);

    #[async_trait::async_trait]
    impl SettlementSink for RecordingSink {
        async fn persist_settlement(&self, amount: u128) {
            self.0
                .lock()
                .expect("recording-sink mutex poisoned")
                .push(amount);
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn cancelled_after_settlement_persists_durable_receipt() {
        // #168(b) composes with the settlement-point commit: a turn-deadline drop
        // AFTER settlement but INSIDE fulfill_with charges the in-memory budget
        // (proven above) AND must persist a durable receipt for the same spend —
        // otherwise a restart / second replica, summing only durable receipts,
        // under-counts real onchain spend. The cancellation net persists the
        // amount the moment the future is dropped, so the durable record survives
        // even though no `.await` runs in a drop.
        let server = MockServer::start().await;
        mount_402_then_hang(&server).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let sink = RecordingSink::default();
        let cap = SpendCap::unlimited();
        let mut call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        call.settlement_sink = Some(std::sync::Arc::new(sink.clone()));
        let fut = fulfill_with(call, &PayingStub, &spend);
        // Drop the future while the authed retry hangs (a turn-deadline drop between
        // settlement and the commit/return).
        let out = tokio::time::timeout(std::time::Duration::from_millis(300), fut).await;
        assert!(
            out.is_err(),
            "the hung retry must outlast the deadline so the future is dropped mid-settlement"
        );
        // The detached persist runs on the runtime after the drop; poll for it.
        let mut recorded = None;
        for _ in 0..80 {
            if let Some(&first) = sink.0.lock().unwrap().first() {
                recorded = Some(first);
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        assert_eq!(
            recorded,
            Some(10_000),
            "a drop after settlement must record the charged spend durably (no lost charge)"
        );
        // The reservation was taken at the per-call ceiling and is still held:
        // the caller's own commit is what narrows it to the charged figure, and
        // this drop never reached one. Over-counting an in-flight charge is the
        // fail-closed direction.
        assert_eq!(spend.held("conv1"), 100_000);
    }

    /// Mounts a `/paid` endpoint that charges TWICE in one call: the unauthed
    /// probe gets challenge A, the retry carrying credential A gets a SECOND 402
    /// (challenge B, a different id and a larger amount), and only a retry
    /// carrying a second credential would be served the resource.
    ///
    /// This is the incremental-payment shape the 402 exchange is willing to pay:
    /// it pays up to `DEFAULT_MAX_PAYMENT_RETRIES` distinct challenges per call
    /// and only refuses to re-pay a challenge id it already paid.
    async fn mount_402_then_402_then_200(server: &MockServer) {
        // Served exactly once, so the SECOND authed retry (if one were ever made)
        // falls through to the 200 below rather than looping on another 402.
        Mock::given(method("GET"))
            .and(path("/paid"))
            .and(header_exists("authorization"))
            .respond_with(ResponseTemplate::new(402).insert_header(
                "www-authenticate",
                challenge_header_for("multi-b", "20000").as_str(),
            ))
            .up_to_n_times(1)
            .with_priority(1)
            .mount(server)
            .await;
        // The merchant's success path. Reaching it means a SECOND credential was
        // handed over — the money bug this test exists to keep out.
        Mock::given(method("GET"))
            .and(path("/paid"))
            .and(header_exists("authorization"))
            .respond_with(ResponseTemplate::new(200).set_body_string("paid-body"))
            .with_priority(2)
            .mount(server)
            .await;
        Mock::given(method("GET"))
            .and(path("/paid"))
            .respond_with(ResponseTemplate::new(402).insert_header(
                "www-authenticate",
                challenge_header_for("multi-a", "10000").as_str(),
            ))
            .with_priority(5)
            .mount(server)
            .await;
    }

    #[tokio::test]
    async fn refuses_a_second_payment_in_one_call() {
        // The multi-payment 402 path, and the reason the single-use reservation
        // is a money fix rather than only a cancellation fix.
        //
        // The exchange pays distinct challenges up to its retry limit, so
        // 402(A, 10000) → pay A → 402(B, 20000) → pay B → 200 is a sequence one
        // call can walk. Committing the charge after the call returned counted
        // ONE of those two payments while both moved money: a silent under-count,
        // and a license for one call to spend several times its own cap.
        //
        // Committing at the settlement point makes the second payment impossible
        // instead of merely miscounted. The reservation is single-use, so B's
        // commit is refused and B's credential — already minted — is discarded
        // rather than sent. The merchant keeps A's payment and does not serve the
        // resource: the stated trade in the module docs' "one call, one charge"
        // section.
        //
        // Anyone later "fixing" incremental payments by re-arming the reservation
        // must make this test fail first.
        let server = MockServer::start().await;
        mount_402_then_402_then_200(&server).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let provider = CountingPayStub::default();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", 100_000, &cap),
            &provider,
            &spend,
        )
        .await;

        assert!(
            !out.is_ok(),
            "the refused second payment must fail the call, not quietly succeed"
        );
        assert!(
            matches!(
                out.reason(),
                Some(RejectReason::Payment(PaymentError::Verification(_)))
            ),
            "expected the refused commit to surface as a verification failure, got {:?}",
            out.reason()
        );
        // The record carries EXACTLY the first charge — not both (the under-count
        // the old ordering produced would have shown 10000 while 30000 moved) and
        // not zero (A really did settle).
        assert_eq!(
            charged(&out),
            Some(10_000),
            "the attribution reports the one payment that was actually made"
        );
        // Credential B was minted and then thrown away: the provider signed
        // twice, the merchant received one credential. That distinction is the
        // whole claim — the refusal happens after the mint and before the send,
        // which is why it costs nothing.
        assert_eq!(
            provider.minted(),
            2,
            "the provider is asked to sign both challenges"
        );
        let requests = server
            .received_requests()
            .await
            .expect("the mock server records requests");
        assert_eq!(
            requests.len(),
            2,
            "only the unauthed probe and challenge A's retry reach the merchant; \
             a third request would mean credential B was sent"
        );
        assert_eq!(
            requests
                .iter()
                .filter(|r| r.headers.contains_key("authorization"))
                .count(),
            1,
            "exactly one credential is handed to the merchant"
        );
    }

    #[tokio::test]
    async fn payment_failure_releases_budget() {
        // A 402 the NoopProvider cannot satisfy → paid_get_with errors → the
        // reservation must be released (budget back to 0).
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .and(path("/paid"))
            .respond_with(
                ResponseTemplate::new(402).insert_header("www-authenticate", "Payment realm=\"x\""),
            )
            .mount(&server)
            .await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", 100_000, &cap),
            &NoopProvider,
            &spend,
        )
        .await;
        assert!(!out.is_ok());
        assert!(matches!(out.reason(), Some(RejectReason::Payment(_))));
        assert!(
            out.paid().is_none(),
            "a failed payment must record no charge"
        );
    }

    /// An unreachable spend authority refuses the payment rather than reporting
    /// a zero floor.
    ///
    /// This is the fail-closed flip C3.1 makes. The floor this replaced answered
    /// zero on any read failure, which is not a conservative default: zero means
    /// "nothing spent", so a conversation that had already spent its ceiling got
    /// the whole ceiling back every time the record could not be read. Refusing
    /// costs one paid call; reporting zero costs the cap, repeatedly.
    #[tokio::test]
    async fn an_unreachable_spend_authority_refuses_rather_than_reporting_zero() {
        let server = MockServer::start().await;
        mount_402_then_200(&server, true).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(1_000_000);
        spend.go_dark();
        let cap = SpendCap::unlimited();
        let call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        let held = call.reservation.clone();
        let (out, _net) = fulfill_with(call, &PayingStub, &spend).await;
        assert!(
            matches!(out.reason(), Some(RejectReason::SpendUnreachable)),
            "an unreachable authority must refuse, got {:?}",
            out.reason()
        );
        assert!(out.paid().is_none(), "the refused call paid nothing");
        assert_eq!(
            held.settlement(),
            None,
            "a call refused before signing resolves nothing"
        );
    }

    /// A conversation at its ceiling and an authority that cannot answer are
    /// different refusals, so the two never share wording.
    #[tokio::test]
    async fn an_over_ceiling_refusal_is_not_worded_as_an_unreachable_one() {
        let server = MockServer::start().await;
        mount_402_then_200(&server, true).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        // A ceiling of 100 with a per-call reservation of 100_000 refuses on the
        // ceiling, with the authority perfectly reachable.
        let spend = Ceiling::new(100);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", 100_000, &cap),
            &PayingStub,
            &spend,
        )
        .await;
        assert!(
            matches!(out.reason(), Some(RejectReason::OverBudget(_))),
            "a conversation at its ceiling must not read as an unreachable authority, got {:?}",
            out.reason()
        );
    }

    /// Moderato pathUSD — a known 6-decimal currency, so the `CappedProvider`
    /// amount guard engages (rather than failing on unknown decimals).
    const PATH_USD: &str = "0x20c0000000000000000000000000000000000000";

    /// Mounts a 402-only endpoint at `/paid` carrying a hostile challenge built
    /// from `request_json`. The provider's `pay` rejects it pre-signing, so the
    /// authed retry is never made (no 200 mock needed).
    async fn mount_402_with_request(server: &MockServer, request_json: serde_json::Value) {
        let request = Base64UrlJson::from_value(&request_json).unwrap();
        let challenge =
            PaymentChallenge::new("proxy-mal", "polychrome", "tempo", "charge", request);
        let header = format_www_authenticate(&challenge).unwrap();
        Mock::given(method("GET"))
            .and(path("/paid"))
            .respond_with(
                ResponseTemplate::new(402).insert_header("www-authenticate", header.as_str()),
            )
            .mount(server)
            .await;
    }

    #[tokio::test]
    async fn over_cap_challenge_does_not_consume_budget() {
        // A malicious 402 demands MORE than the per-call cap. Wrapped in a real
        // CappedProvider, `pay` is rejected BEFORE any credential is created — so
        // budget must NOT commit and no attribution may ride (no false spend
        // event).
        let server = MockServer::start().await;
        // amount 200000 base units = 0.20 pathUSD > the 0.10 CappedProvider cap.
        mount_402_with_request(
            &server,
            serde_json::json!({ "amount": "200000", "currency": PATH_USD }),
        )
        .await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        // PayingStub WOULD pay, but the CappedProvider gates it first.
        let capped = CappedProvider::new(PayingStub, "0.10", PATH_USD, 6).expect("cap parses");
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", 100_000, &cap),
            &capped,
            &spend,
        )
        .await;
        assert!(!out.is_ok(), "an over-cap challenge must be rejected");
        assert!(
            out.paid().is_none(),
            "no credential was created → must NOT flag a payment"
        );
        assert!(
            out.paid().is_none(),
            "a rejected over-cap challenge must record no charge (no false spend)"
        );
    }

    #[tokio::test]
    async fn wrong_chain_challenge_does_not_consume_budget() {
        // A malicious 402 steers onto the wrong chain. The CappedProvider's chain
        // guard rejects it before signing — no credential, no budget consumed.
        let server = MockServer::start().await;
        mount_402_with_request(
            &server,
            serde_json::json!({
                "amount": "10000", // 0.01 — well within cap
                "currency": PATH_USD,
                "methodDetails": { "chainId": 1 } // Ethereum mainnet, not Moderato
            }),
        )
        .await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let capped = CappedProvider::new(PayingStub, "1.00", PATH_USD, 6)
            .expect("cap parses")
            .with_expected_chain_id(42431); // deployment network: Moderato
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", 100_000, &cap),
            &capped,
            &spend,
        )
        .await;
        assert!(!out.is_ok(), "a wrong-chain challenge must be rejected");
        assert!(
            out.paid().is_none(),
            "no credential was created → must NOT flag a payment"
        );
        assert!(
            out.paid().is_none(),
            "a rejected wrong-chain challenge must record no charge (no false spend)"
        );
    }

    #[tokio::test]
    async fn rejects_over_presign_per_tx_cap_before_signing() {
        // Pre-authorization (pre-signing spend cap): a request whose per-call
        // ceiling exceeds the granted per-transaction mandate is rejected BEFORE
        // the budget is reserved and BEFORE the 402 exchange — no credential, no
        // spend. PayingStub WOULD pay, and the mock WOULD serve, so a rejection
        // proves the pre-auth gate (not a missing server) blocked it.
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200).set_body_string("unreached"))
            .mount(&server)
            .await;
        let url = format!("{}/x", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(10_000_000);
        let cap = SpendCap::new(Some(50_000), None); // pre-authorized $0.05 per tx
        let (out, _net) = fulfill_with(
            basic_call(
                &a, &a, &url, "conv1",
                100_000, // requested per-call ceiling $0.10 > the $0.05 pre-auth
                &cap,
            ),
            &PayingStub,
            &spend,
        )
        .await;
        assert!(!out.is_ok(), "an over-pre-auth request must be rejected");
        assert!(
            matches!(out.reason(), Some(RejectReason::OverSpendCap(_))),
            "rejection must name the pre-authorization cap: {:?}",
            out.reason()
        );
        assert!(
            out.paid().is_none(),
            "no payment was made (no credential minted)"
        );
        assert_eq!(
            spend.held("conv1"),
            0,
            "a pre-auth rejection reserves no budget"
        );
    }

    #[tokio::test]
    async fn rejects_over_presign_per_session_cap_before_signing() {
        // The per-session pre-authorization bounds the running total: with $0.40
        // already committed this session and a $0.45 session mandate, a further
        // $0.10 call breaches the session ceiling and is rejected before signing.
        let server = MockServer::start().await;
        Mock::given(method("GET"))
            .respond_with(ResponseTemplate::new(200).set_body_string("unreached"))
            .mount(&server)
            .await;
        let url = format!("{}/x", server.uri());
        let a = args(&url);

        let spend = Ceiling::new(10_000_000);
        // $0.40 already spent this session, durably.
        spend
            .reserve("conv1", "earlier-call", 400_000)
            .await
            .expect("the earlier call fit");
        let cap = SpendCap::new(None, Some(450_000)); // $0.45 per-session mandate
        let (out, _net) = fulfill_with(
            basic_call(
                &a, &a, &url, "conv1",
                100_000, // 400_000 + 100_000 = 500_000 > the 450_000 session cap
                &cap,
            ),
            &PayingStub,
            &spend,
        )
        .await;
        assert!(!out.is_ok(), "an over-session-cap request must be rejected");
        assert!(matches!(out.reason(), Some(RejectReason::OverSpendCap(_))));
        assert_eq!(
            spend.held("conv1"),
            400_000,
            "the rejected call adds nothing; only the prior $0.40 stays committed"
        );
    }

    /// One-command live Moderato smoke of the WHOLE proxy chokepoint with a real
    /// keychain key — resolve (keychain + atomic provisioning) → `CappedProvider`
    /// (cap + chain guard) → SSRF → budget reserve → 402 pay/retry → commit. No
    /// LLM / harness orchestration. Run after `tempo wallet login -n moderato`
    /// + `tempo wallet -n moderato fund`, against a Moderato 402 endpoint
    /// (`tempo wallet -n moderato -t services`):
    ///
    /// ```sh
    /// TEMPO_WALLET_KEYS_PATH=$HOME/.tempo/wallet/keys.toml \
    /// TEMPO_CURRENCY=0x20c0000000000000000000000000000000000000 \
    /// TEMPO_CHAIN_ID=42431 \
    /// TEMPO_TEST_PAID_URL=<a moderato 402 url> \
    /// cargo test -p polyc-payments live_moderato_keychain_payment -- --ignored --nocapture
    /// ```
    #[tokio::test]
    #[ignore = "requires a funded Moderato keychain keys.toml + TEMPO_TEST_PAID_URL"]
    async fn live_moderato_keychain_payment() {
        use crate::config::PaymentsConfig;
        let cfg = PaymentsConfig::from_env()
            .expect("payments env (TEMPO_WALLET_KEYS_PATH + TEMPO_CURRENCY + TEMPO_CHAIN_ID)");
        let url = std::env::var("TEMPO_TEST_PAID_URL")
            .expect("set TEMPO_TEST_PAID_URL to a Moderato 402 endpoint");
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_secs();
        // Generous per-conversation budget + per-call cap for the smoke.
        let spend = Ceiling::new(
            crate::amount::dollars_to_base_units("10.00", cfg.currency_decimals).unwrap(),
        );
        let args = format!(r#"{{"url":"{url}","max_spend":"1.00"}}"#);
        // approved_args == args (we self-approve in the smoke; the byte-match and
        // tool-name binding are covered by the hermetic control-plane tests).
        let (resp, _net) = fulfill(
            FulfillRequest {
                args_json: &args,
                approved_args_json: &args,
                url: &url,
                method: reqwest::Method::GET,
                body: None,
                headers: &[],
                max_spend: Some("1.00"),
                key_source: KeySource::FileSecret,
                conversation_id: "live-conv",
                reservation_id: "res-1",
                now_unix: now,
                mandate_chain: None,
                recorder: None,
                settlement_sink: None,
                persona_per_payment_limit: None,
                persona_allowed_hosts: None,
            },
            &cfg,
            &spend,
        )
        .await;
        eprintln!(
            "live keychain payment → status={:?} paid={:?} receipt={:?} reason={:?}",
            resp.status(),
            resp.paid(),
            resp.receipt(),
            resp.reason()
        );
        assert!(resp.is_ok(), "keychain payment failed: {:?}", resp.reason());
        assert!(
            resp.paid().is_some(),
            "expected an onchain payment (pay invoked)"
        );
    }

    /// Sign and chain-verify a mandate for `host` / `amount` (base-unit
    /// decimal string) / `args_json`, bound to conversation `conv1` — the
    /// mandate-side fixture for the enforcement-seam tests below.
    fn verified_mandate_for(
        host: &str,
        amount: &str,
        args_json: &str,
    ) -> polyc_crypto::mandate::VerifiedMandateChain {
        use polyc_crypto::Signer;
        use polyc_crypto::mandate::{
            CartFields, IntentFields, PaymentFields, mandate_hash, sign_cart_mandate,
            sign_intent_mandate, sign_payment_mandate,
        };
        let signer = Signer::from_seed(77);
        let (intent, _s, _p) = sign_intent_mandate(
            &IntentFields {
                caller: "slack:T1:U9",
                conversation_id: "conv1",
                scope_description: "proxy seam test",
                currency: PATH_USD,
                max_total_base_units: "",
                issued_at_unix: 0,
                expires_at_unix: u64::MAX,
                nonce: "n-intent",
            },
            &signer,
        );
        let (cart, _s, _p) = sign_cart_mandate(
            &CartFields {
                intent_hash: &mandate_hash(&intent),
                caller: "slack:T1:U9",
                conversation_id: "conv1",
                merchant_host: host,
                currency: PATH_USD,
                amount_base_units: amount,
                issued_at_unix: 0,
                expires_at_unix: u64::MAX,
                nonce: "n-cart",
            },
            &signer,
        );
        let (payment, _s, _p) = sign_payment_mandate(
            &PaymentFields {
                cart_hash: &mandate_hash(&cart),
                caller: "slack:T1:U9",
                conversation_id: "conv1",
                args_json,
                currency: PATH_USD,
                amount_base_units: amount,
                issued_at_unix: 0,
                expires_at_unix: u64::MAX,
                nonce: "n-payment",
            },
            &signer,
        );
        polyc_crypto::mandate::MandateChain {
            intent,
            cart,
            payment,
        }
        .verify("conv1", &signer.public_key_bytes(), 1_000)
        .expect("fixture chain verifies")
    }

    #[tokio::test]
    async fn valid_mandate_authorizes_the_payment() {
        // Enforcement seam, positive half: a verified chain covering the
        // amount, the destination host, and the approved args admits the
        // payment — the 402 exchange settles and budget commits exactly as
        // the mandate-less path does.
        let server = MockServer::start().await;
        mount_402_then_200(&server, true).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);
        let host = reqwest::Url::parse(&server.uri())
            .unwrap()
            .host_str()
            .unwrap()
            .to_owned();

        let mandate = verified_mandate_for(&host, "100000", &a);
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let mut call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        call.gates.mandate = Some(&mandate);
        let (out, _net) = fulfill_with(call, &PayingStub, &spend).await;
        assert!(out.is_ok(), "expected success, got {:?}", out.reason());
        assert!(
            out.paid().is_some(),
            "the mandate-authorized payment settles"
        );
        assert_eq!(
            charged(&out),
            Some(10_000),
            "the charged amount is what gets recorded"
        );
    }

    #[tokio::test]
    async fn mandate_below_requested_spend_refuses_before_signing() {
        // The chain authorizes LESS than the per-call ceiling: refused at the
        // pre-sign seam — no credential, no reservation, nothing fetched.
        let server = MockServer::start().await;
        mount_402_then_200(&server, true).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);
        let host = reqwest::Url::parse(&server.uri())
            .unwrap()
            .host_str()
            .unwrap()
            .to_owned();

        let mandate = verified_mandate_for(&host, "50000", &a); // < the 100_000 ceiling
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let mut call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        call.gates.mandate = Some(&mandate);
        let (out, _net) = fulfill_with(call, &PayingStub, &spend).await;
        assert!(!out.is_ok(), "an under-authorized mandate must refuse");
        assert!(matches!(
            out.reason(),
            Some(RejectReason::MandateRefused(_))
        ));
        assert!(out.paid().is_none(), "no credential was minted");
        assert_eq!(spend.held("conv1"), 0, "no reservation was made");
    }

    #[tokio::test]
    async fn mandate_for_a_different_host_refuses_before_signing() {
        // The chain is scoped to a DIFFERENT merchant than the approved URL
        // targets: refused at the pre-sign seam even though the amount fits.
        let server = MockServer::start().await;
        mount_402_then_200(&server, true).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        let mandate = verified_mandate_for("api.other-merchant.com", "100000", &a);
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let mut call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        call.gates.mandate = Some(&mandate);
        let (out, _net) = fulfill_with(call, &PayingStub, &spend).await;
        assert!(!out.is_ok(), "a wrong-merchant mandate must refuse");
        assert!(matches!(
            out.reason(),
            Some(RejectReason::MandateRefused(_))
        ));
        assert_eq!(spend.held("conv1"), 0);
    }

    #[tokio::test]
    async fn mandate_bound_to_different_args_refuses_before_signing() {
        // The Payment link binds the EXACT approved args; a mandate minted
        // for another call cannot authorize this one.
        let server = MockServer::start().await;
        mount_402_then_200(&server, true).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);
        let host = reqwest::Url::parse(&server.uri())
            .unwrap()
            .host_str()
            .unwrap()
            .to_owned();

        let mandate = verified_mandate_for(
            &host,
            "100000",
            r#"{"url":"https://api.example.com/other"}"#,
        );
        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let mut call = basic_call(&a, &a, &url, "conv1", 100_000, &cap);
        call.gates.mandate = Some(&mandate);
        let (out, _net) = fulfill_with(call, &PayingStub, &spend).await;
        assert!(!out.is_ok(), "an args-mismatched mandate must refuse");
        assert!(matches!(
            out.reason(),
            Some(RejectReason::MandateRefused(_))
        ));
        assert_eq!(spend.held("conv1"), 0);
    }

    #[tokio::test]
    async fn fulfill_ignores_presented_chain_when_mandates_unconfigured() {
        // Feature gate, production wrapper: with NO issuer key configured
        // (the default), a presented chain — even a garbage one — is ignored
        // and behavior is exactly today's path. The call proceeds past the
        // mandate seam and fails later at signer resolution
        // (BackendUnavailable), proving the chain neither authorized nor
        // refused anything.
        let cfg = PaymentsConfig::from_lookup(|k| match k {
            "TEMPO_CURRENCY" => Some(PATH_USD.to_owned()),
            _ => None,
        })
        .expect("config builds");
        assert!(cfg.mandate_issuer_public_key().is_none(), "off by default");
        let garbage = MandateChain {
            intent: b"not a mandate".to_vec(),
            cart: b"not a mandate".to_vec(),
            payment: b"not a mandate".to_vec(),
        };
        let spend = Ceiling::new(1_000_000);
        let a = args("https://api.example.com/x");
        let (out, _net) = fulfill(
            FulfillRequest {
                args_json: &a,
                approved_args_json: &a,
                url: "https://api.example.com/x",
                method: reqwest::Method::GET,
                body: None,
                headers: &[],
                max_spend: None,
                key_source: KeySource::FileSecret,
                conversation_id: "conv1",
                reservation_id: "res-1",
                now_unix: 1_000,
                mandate_chain: Some(&garbage),
                recorder: None,
                settlement_sink: None,
                persona_per_payment_limit: None,
                persona_allowed_hosts: None,
            },
            &cfg,
            &spend,
        )
        .await;
        assert!(
            matches!(out.reason(), Some(RejectReason::BackendUnavailable(_))),
            "must fail at signer resolution (past the mandate seam), got: {:?}",
            out.reason()
        );
    }

    #[tokio::test]
    async fn fulfill_refuses_invalid_chain_when_mandates_configured() {
        // Feature gate, other half: WITH an issuer key configured, a
        // presented-but-invalid chain refuses the payment outright — before
        // signer resolution (fail closed, never silently ignored).
        let issuer_hex = hex::encode(polyc_crypto::Signer::from_seed(77).public_key_bytes());
        let cfg = PaymentsConfig::from_lookup(|k| match k {
            "TEMPO_CURRENCY" => Some(PATH_USD.to_owned()),
            "TEMPO_MANDATE_ISSUER_PUBKEY" => Some(issuer_hex.clone()),
            _ => None,
        })
        .expect("config builds");
        assert!(cfg.mandate_issuer_public_key().is_some());
        let garbage = MandateChain {
            intent: b"not a mandate".to_vec(),
            cart: b"not a mandate".to_vec(),
            payment: b"not a mandate".to_vec(),
        };
        let spend = Ceiling::new(1_000_000);
        let a = args("https://api.example.com/x");
        let (out, _net) = fulfill(
            FulfillRequest {
                args_json: &a,
                approved_args_json: &a,
                url: "https://api.example.com/x",
                method: reqwest::Method::GET,
                body: None,
                headers: &[],
                max_spend: None,
                key_source: KeySource::FileSecret,
                conversation_id: "conv1",
                reservation_id: "res-1",
                now_unix: 1_000,
                mandate_chain: Some(&garbage),
                recorder: None,
                settlement_sink: None,
                persona_per_payment_limit: None,
                persona_allowed_hosts: None,
            },
            &cfg,
            &spend,
        )
        .await;
        assert!(
            matches!(out.reason(), Some(RejectReason::MandateRefused(_))),
            "the refusal must name the mandate gate, got: {:?}",
            out.reason()
        );
        assert_eq!(spend.held("conv1"), 0);
    }

    // ---- Persona spend-policy narrowing (`#514`/`#517`): `fulfill`'s host
    // ---- checks run BEFORE signer resolution, so these are reachable with no
    // ---- live signer configured (an unconfigured `KeySource::FileSecret`
    // ---- would otherwise fail at `BackendUnavailable`, past this seam).

    #[tokio::test]
    async fn paid_fetch_refuses_a_host_off_the_persona_allowlist() {
        // No deployment allowlist configured (only the SSRF guard would apply
        // there) + a persona allowlist naming a DIFFERENT host → refused before
        // ever reaching signer resolution.
        let cfg = PaymentsConfig::from_lookup(|k| match k {
            "TEMPO_CURRENCY" => Some(PATH_USD.to_owned()),
            _ => None,
        })
        .expect("config builds");
        assert!(cfg.paid_host_allowlist.is_none(), "no deployment allowlist");
        let spend = Ceiling::new(1_000_000);
        let a = args("https://evil.example.com/x");
        let persona_hosts = vec!["api.example.com".to_owned()];
        let (out, _net) = fulfill(
            FulfillRequest {
                args_json: &a,
                approved_args_json: &a,
                url: "https://evil.example.com/x",
                method: reqwest::Method::GET,
                body: None,
                headers: &[],
                max_spend: None,
                key_source: KeySource::FileSecret,
                conversation_id: "conv1",
                reservation_id: "res-1",
                now_unix: 1_000,
                mandate_chain: None,
                recorder: None,
                settlement_sink: None,
                persona_per_payment_limit: None,
                persona_allowed_hosts: Some(&persona_hosts),
            },
            &cfg,
            &spend,
        )
        .await;
        assert!(
            matches!(out.reason(), Some(RejectReason::PersonaHostNotAllowed)),
            "must refuse via the persona-allowlist gate, got: {:?}",
            out.reason()
        );
        assert_eq!(spend.held("conv1"), 0);
    }

    #[tokio::test]
    async fn persona_allowlist_narrows_not_replaces_the_deployment_allowlist() {
        // A host allowed by the DEPLOYMENT allowlist but NOT on the persona's
        // own list is still refused — the persona list narrows, it never
        // widens or replaces, the deployment's own bound.
        let cfg = PaymentsConfig::from_lookup(|k| match k {
            "TEMPO_CURRENCY" => Some(PATH_USD.to_owned()),
            "TEMPO_PAID_HOST_ALLOWLIST" => Some("api.example.com,other.example.com".to_owned()),
            _ => None,
        })
        .expect("config builds");
        assert_eq!(
            cfg.paid_host_allowlist.as_deref(),
            Some(&["api.example.com".to_owned(), "other.example.com".to_owned()][..])
        );
        let spend = Ceiling::new(1_000_000);
        // "api.example.com" passes the DEPLOYMENT allowlist...
        let a = args("https://api.example.com/x");
        let persona_hosts = vec!["only-this-host.example.com".to_owned()];
        let (out, _net) = fulfill(
            FulfillRequest {
                args_json: &a,
                approved_args_json: &a,
                url: "https://api.example.com/x",
                method: reqwest::Method::GET,
                body: None,
                headers: &[],
                max_spend: None,
                key_source: KeySource::FileSecret,
                conversation_id: "conv1",
                reservation_id: "res-1",
                now_unix: 1_000,
                mandate_chain: None,
                recorder: None,
                settlement_sink: None,
                persona_per_payment_limit: None,
                // ...but is not on the CALLER's own persona allowlist.
                persona_allowed_hosts: Some(&persona_hosts),
            },
            &cfg,
            &spend,
        )
        .await;
        assert!(
            matches!(out.reason(), Some(RejectReason::PersonaHostNotAllowed)),
            "a deployment-allowed host must still be refused when it's off the \
             caller's OWN persona allowlist, got: {:?}",
            out.reason()
        );
        assert_eq!(spend.held("conv1"), 0);

        // An EMPTY persona allowlist (unset/cleared policy) never narrows —
        // the same host now passes with no persona list at all.
        let (out_no_persona_list, _net) = fulfill(
            FulfillRequest {
                args_json: &a,
                approved_args_json: &a,
                url: "https://api.example.com/x",
                method: reqwest::Method::GET,
                body: None,
                headers: &[],
                max_spend: None,
                key_source: KeySource::FileSecret,
                conversation_id: "conv1",
                reservation_id: "res-1",
                now_unix: 1_000,
                mandate_chain: None,
                recorder: None,
                settlement_sink: None,
                persona_per_payment_limit: None,
                persona_allowed_hosts: None,
            },
            &cfg,
            &spend,
        )
        .await;
        assert!(
            !matches!(
                out_no_persona_list.reason(),
                Some(RejectReason::PersonaHostNotAllowed)
            ),
            "no persona allowlist must not itself refuse a deployment-allowed host, got: {:?}",
            out_no_persona_list.reason()
        );
    }

    // A challenge amount above the caller's OWN persona per-payment ceiling
    // but below the deployment default cap is refused, and the budget
    // reservation is released — the SAME `CappedProvider` enforcement
    // `fulfill` builds from the fold-narrowed cap
    // (`CappedProvider::from_base_units`), proven directly here without a
    // live signer.
    #[tokio::test]
    async fn persona_per_payment_limit_caps_the_payment_below_the_deployment_cap() {
        let server = MockServer::start().await;
        // 0.15 pathUSD: above the persona's 0.10 cap, but below the
        // deployment's 1.00 default — must be refused only because of the
        // NARROWED (persona) cap, not the deployment one.
        mount_402_with_request(
            &server,
            serde_json::json!({ "amount": "150000", "currency": PATH_USD }),
        )
        .await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);

        // Mirrors exactly what `fulfill` computes: deployment cap "1.00" (6
        // decimals) folded with a caller persona_per_payment_limit of "0.10"
        // → the SMALLER wins.
        let deployment_cap_base_units = 1_000_000u128; // "1.00"
        let persona_cap_base_units = 100_000u128; // "0.10"
        let narrowed = persona_cap_base_units.min(deployment_cap_base_units);
        assert_eq!(narrowed, persona_cap_base_units, "the persona cap must win");
        let capped =
            CappedProvider::from_base_units(PayingStub, narrowed, PATH_USD).expect("cap parses");

        let spend = Ceiling::new(1_000_000);
        let cap = SpendCap::unlimited();
        let (out, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", narrowed, &cap),
            &capped,
            &spend,
        )
        .await;
        assert!(
            !out.is_ok(),
            "a challenge above the persona-narrowed cap must be refused"
        );
        assert!(
            out.paid().is_none(),
            "no credential was created → must NOT flag a payment"
        );
        assert!(
            out.paid().is_none(),
            "a rejected over-persona-cap challenge must record no charge — the SAME \
             untouched-on-refusal guarantee the deployment cap already carries"
        );
    }

    // Issue #2365: the mirror that used to seed this fold from a chain figure
    // (the removed #2246 sync RPC, `ForensicsService.SyncSpendingLimit`) is
    // gone — the ceiling is set ONLY by a deliberate admin action
    // (`wallet_nav::set_policy`), so it is never
    // stale by construction. This proves the availability goal the removed
    // mirror was chasing still holds with no chain read anywhere in this
    // path: raising the STORED string a caller's admin chose, with no
    // ceremony beyond `wallet_set_policy`, immediately raises what the next
    // payment decision permits — end to end, not merely at the fold.
    #[tokio::test]
    async fn persona_per_payment_limit_raised_by_the_admin_permits_a_payment_the_old_value_refused()
    {
        let server = MockServer::start().await;
        // The standard $0.01 (10,000 base unit) 402-then-200 challenge — well
        // below the deployment's $1.00 default cap, so only the persona
        // per-payment ceiling decides whether either attempt is refused or
        // allowed.
        mount_402_then_200(&server, false).await;
        let url = format!("{}/paid", server.uri());
        let a = args(&url);
        let deployment_cap_base_units = 1_000_000u128; // "1.00"
        let challenge_base_units = 10_000u128; // "0.01"
        let spend = Ceiling::new(10_000_000);
        let cap = SpendCap::unlimited();

        // BEFORE: the admin's original "0.005" ceiling — below the challenge.
        let original_limit = crate::amount::format_bare_amount(5_000, 6);
        assert_eq!(original_limit, "0.005");
        let original_cap = crate::amount::dollars_to_base_units(&original_limit, 6)
            .expect("parses")
            .min(deployment_cap_base_units);
        assert!(
            original_cap < challenge_base_units,
            "sanity: the original limit is below the challenge"
        );
        let capped_original = CappedProvider::from_base_units(PayingStub, original_cap, PATH_USD)
            .expect("cap parses");
        let (out_original, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv1", original_cap, &cap),
            &capped_original,
            &spend,
        )
        .await;
        assert!(
            !out_original.is_ok(),
            "the $0.01 challenge exceeds the original $0.005 limit and must be refused"
        );

        // AFTER: the admin raises the SAME field, in chat, to "0.02" — no
        // chain write, no sync call, nothing but `wallet_set_policy`.
        let raised_limit = crate::amount::format_bare_amount(20_000, 6);
        assert_eq!(raised_limit, "0.02");
        let raised_cap = crate::amount::dollars_to_base_units(&raised_limit, 6)
            .expect("the raised figure parses")
            .min(deployment_cap_base_units);
        assert!(
            raised_cap > challenge_base_units,
            "sanity: the raised cap is now above the $0.01 challenge"
        );
        let capped_raised =
            CappedProvider::from_base_units(PayingStub, raised_cap, PATH_USD).expect("cap parses");
        let (out_raised, _net) = fulfill_with(
            basic_call(&a, &a, &url, "conv2", raised_cap, &cap),
            &capped_raised,
            &spend,
        )
        .await;
        assert!(
            out_raised.is_ok(),
            "the SAME $0.01 challenge must now be permitted under the admin's \
             raised limit: {:?}",
            out_raised.reason()
        );
    }

    // ---- `TEMPO_MAX_SPEND` is a CEILING (#170), never a default an
    // ---- agent-authored `max_spend` can widen past.

    #[test]
    fn resolve_cap_prefers_the_tighter_of_agent_and_operator_caps() {
        // Operator ceiling "0.10", agent asks for "1.00" (wider) → the
        // operator's ceiling wins, not the agent's wider ask.
        let cap = resolve_cap_base_units(Some("1.00"), Some("0.10"), 6).expect("parses");
        assert_eq!(
            cap, 100_000,
            "operator ceiling must win over a wider agent ask"
        );

        // Agent asks for a NARROWER cap than the operator ceiling → the
        // agent's tighter ask wins (narrowing is always allowed).
        let cap = resolve_cap_base_units(Some("0.05"), Some("0.10"), 6).expect("parses");
        assert_eq!(cap, 50_000, "a narrower agent ask must still win");

        // No operator ceiling configured → the agent's ask alone applies.
        let cap = resolve_cap_base_units(Some("2.50"), None, 6).expect("parses");
        assert_eq!(
            cap, 2_500_000,
            "no operator ceiling ⇒ agent ask alone applies"
        );

        // No agent-authored max_spend → the operator ceiling alone applies.
        let cap = resolve_cap_base_units(None, Some("0.10"), 6).expect("parses");
        assert_eq!(
            cap, 100_000,
            "no agent ask ⇒ operator ceiling alone applies"
        );

        // Neither side supplies a value → DEFAULT_CAP.
        let cap = resolve_cap_base_units(None, None, 6).expect("parses");
        assert_eq!(
            cap,
            crate::amount::dollars_to_base_units(DEFAULT_CAP, 6).unwrap(),
            "neither side set ⇒ DEFAULT_CAP"
        );
    }

    #[test]
    fn resolve_cap_fails_closed_on_an_unparseable_value_rather_than_widening() {
        // An agent-authored `max_spend` that overflows `u128` must REFUSE, not
        // silently fall back to DEFAULT_CAP — DEFAULT_CAP ("0.10") can be
        // WIDER than an operator ceiling the malformed value should have stayed
        // under.
        let overflowing = "999999999999999999999999999999999999999999999999999999";
        let err = resolve_cap_base_units(Some(overflowing), Some("0.01"), 6)
            .expect_err("an overflowing agent max_spend must reject, not fall back");
        assert!(matches!(err, RejectReason::InvalidMaxSpend));

        // Garbage input fails the same way.
        let err = resolve_cap_base_units(Some("not-a-number"), None, 6)
            .expect_err("garbage max_spend must reject");
        assert!(matches!(err, RejectReason::InvalidMaxSpend));

        // An unparseable OPERATOR ceiling must also fail closed — never
        // silently ignored in favor of a wider fallback.
        let err = resolve_cap_base_units(Some("0.01"), Some("not-a-number"), 6)
            .expect_err("an unparseable operator ceiling must reject");
        assert!(matches!(err, RejectReason::InvalidMaxSpend));
    }

    #[tokio::test]
    async fn fulfill_enforces_the_operator_ceiling_over_a_wider_agent_max_spend() {
        // End-to-end through `fulfill`: operator `TEMPO_MAX_SPEND` = "0.01",
        // agent asks for "1.00" (far wider). A challenge for "0.05" — under the
        // agent's ask, but OVER the operator ceiling — must be refused via the
        // enforced (narrower) cap, proving the ceiling actually gates the
        // signer rather than merely being read and ignored.
        let cfg = PaymentsConfig::from_lookup(|k| match k {
            "TEMPO_CURRENCY" => Some(PATH_USD.to_owned()),
            "TEMPO_MAX_SPEND" => Some("0.01".to_owned()),
            _ => None,
        })
        .expect("config builds");
        assert_eq!(cfg.max_spend.as_deref(), Some("0.01"));

        let spend = Ceiling::new(1_000_000);
        let a = args("https://api.example.com/x");
        let (out, _net) = fulfill(
            FulfillRequest {
                args_json: &a,
                approved_args_json: &a,
                url: "https://api.example.com/x",
                method: reqwest::Method::GET,
                body: None,
                headers: &[],
                max_spend: Some("1.00"),
                key_source: KeySource::FileSecret,
                conversation_id: "conv1",
                reservation_id: "res-1",
                now_unix: 1_000,
                mandate_chain: None,
                recorder: None,
                settlement_sink: None,
                persona_per_payment_limit: None,
                persona_allowed_hosts: None,
            },
            &cfg,
            &spend,
        )
        .await;
        // No live signer is configured (`KeySource::FileSecret` with no key on
        // disk), so this reaches `BackendUnavailable` rather than actually
        // signing — proving the call proceeded PAST the cap-resolution seam
        // with the narrower cap folded in (a widen-past-the-ceiling bug would
        // have produced the exact same downstream failure, which is why the
        // pure-function tests above are the direct proof; this test proves the
        // production `fulfill` entry point actually calls through to it rather
        // than the two drifting apart).
        assert!(
            matches!(out.reason(), Some(RejectReason::BackendUnavailable(_))),
            "must proceed past cap resolution to signer resolution, got: {:?}",
            out.reason()
        );
    }

    #[tokio::test]
    async fn fulfill_rejects_rather_than_widens_on_an_unparseable_max_spend() {
        // An agent-authored `max_spend` that fails to parse must refuse
        // (`InvalidMaxSpend`), not silently fall back to `DEFAULT_CAP` — the
        // previous behavior, which this issue (#170) considered a fail-open
        // bug: an operator ceiling narrower than `DEFAULT_CAP` would have been
        // bypassed by an agent sending garbage.
        let cfg = PaymentsConfig::from_lookup(|k| match k {
            "TEMPO_CURRENCY" => Some(PATH_USD.to_owned()),
            "TEMPO_MAX_SPEND" => Some("0.01".to_owned()),
            _ => None,
        })
        .expect("config builds");
        let spend = Ceiling::new(1_000_000);
        let a = args("https://api.example.com/x");
        let (out, _net) = fulfill(
            FulfillRequest {
                args_json: &a,
                approved_args_json: &a,
                url: "https://api.example.com/x",
                method: reqwest::Method::GET,
                body: None,
                headers: &[],
                max_spend: Some("not-a-number"),
                key_source: KeySource::FileSecret,
                conversation_id: "conv1",
                reservation_id: "res-1",
                now_unix: 1_000,
                mandate_chain: None,
                recorder: None,
                settlement_sink: None,
                persona_per_payment_limit: None,
                persona_allowed_hosts: None,
            },
            &cfg,
            &spend,
        )
        .await;
        assert!(
            matches!(out.reason(), Some(RejectReason::InvalidMaxSpend)),
            "an unparseable max_spend must fail closed, got: {:?}",
            out.reason()
        );
        assert_eq!(
            spend.held("conv1"),
            0,
            "a rejected-before-reservation call must not touch the budget"
        );
    }
}