guardian-client 0.18.1

use crate::auth::Auth;
use crate::error::{ClientError, ClientResult};
use crate::keystore::Signer;
use crate::proto::guardian_client::GuardianClient as GuardianGrpcClient;
use crate::proto::{
    AbandonDeltaCandidateRequest, AbandonDeltaCandidateResponse, AuthConfig, ConfigureRequest,
    ConfigureResponse, GetAccountByKeyCommitmentRequest, GetAccountByKeyCommitmentResponse,
    GetCanonicalNonceRequest, GetCanonicalNonceResponse, GetDeltaHistoryRequest,
    GetDeltaHistoryResponse, GetDeltaProposalRequest, GetDeltaProposalResponse,
    GetDeltaProposalsRequest, GetDeltaProposalsResponse, GetDeltaRequest, GetDeltaResponse,
    GetDeltaSinceRequest, GetDeltaSinceResponse, GetPubkeyRequest, GetStateRequest,
    GetStateResponse, ProposalSignature as ProtoProposalSignature, PushDeltaProposalRequest,
    PushDeltaProposalResponse, PushDeltaRequest, PushDeltaResponse, RevokeAllSessionsRequest,
    RevokeSessionRequest, SessionGrantFields, SignDeltaProposalRequest, SignDeltaProposalResponse,
};
use chrono::Utc;
use guardian_shared::ProposalSignature as JsonProposalSignature;
use guardian_shared::auth_request_message::AuthRequestMessage;
use guardian_shared::auth_request_payload::AuthRequestPayload;
use guardian_shared::hex::{FromHex, IntoHex};
use guardian_shared::lookup_auth_message::LookupAuthMessage;
use guardian_shared::session_grant::{SessionGrant, SessionLogoutMessage, SessionRevokeAllMessage};
use guardian_shared::session_key::SessionKey;
use miden_protocol::Word;
use miden_protocol::account::AccountId;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration;
use tonic::metadata::MetadataValue;
use tonic::transport::Channel;

/// Bounded retry policy for replay-classified rejections
/// (`authentication_replay`): the request was correctly signed but lost the
/// server's per-signer timestamp CAS to a concurrent request. Each attempt
/// re-mints the timestamp and re-signs the identical payload. Mirrors the TS
/// client's policy; terminal authentication failures are never retried.
const REPLAY_RETRY_LIMIT: u32 = 2;
const REPLAY_RETRY_BACKOFF: Duration = Duration::from_millis(50);

/// A client for interacting with Guardian servers.
///
/// `GuardianClient` provides methods for managing off-chain account state, including:
/// - Account configuration
/// - Delta (state change) management
/// - Multi-party proposal workflows
///
/// All methods that interact with account data require authentication via a configured signer.
pub struct GuardianClient {
    client: GuardianGrpcClient<Channel>,
    auth: Option<Auth>,
    signer: Option<Arc<dyn Signer>>,
    session: SessionSlot,
    last_timestamp: AtomicI64,
}

/// Parameters of a session grant the wallet signs.
#[derive(Debug, Clone)]
pub struct StartSessionOptions {
    /// The Guardian's Miden network label: `local`, `devnet` or `testnet`.
    pub network: String,
    /// Session lifetime: more than [`MIN_SESSION_TTL`] and at most the
    /// Guardian's configured maximum (8 hours by default).
    pub ttl: Duration,
}

/// Shortest lifetime Guardian accepts: longer than its 5-minute request
/// clock-skew window.
pub const MIN_SESSION_TTL: Duration = Duration::from_secs(300);
/// Longest lifetime any Guardian accepts.
pub const MAX_SESSION_TTL: Duration = Duration::from_secs(8 * 60 * 60);

/// A registered session.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionInfo {
    pub signer_commitment: String,
    pub session_public_key: String,
    /// Unix seconds.
    pub expires_at: u64,
}

struct ActiveSession {
    key: SessionKey,
    info: SessionInfo,
}

/// Holds a client's session. Clones share it: clients created per operation
/// with [`GuardianClient::with_session_slot`] use one session, and a session
/// that one of them starts, ends or drops is gone for all of them.
#[derive(Clone, Default)]
pub struct SessionSlot(Arc<Mutex<Option<ActiveSession>>>);

impl SessionSlot {
    /// The session, if one is registered and not about to expire.
    pub fn info(&self) -> Option<SessionInfo> {
        let now = Utc::now().timestamp().max(0) as u64;
        self.lock()
            .as_ref()
            .filter(|session| now + SESSION_EXPIRY_MARGIN_SECS < session.info.expires_at)
            .map(|session| session.info.clone())
    }

    fn lock(&self) -> MutexGuard<'_, Option<ActiveSession>> {
        self.0.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Forgets the session if `ended` says so.
    fn clear_if(&self, ended: impl FnOnce(&ActiveSession) -> bool) {
        let mut session = self.lock();
        if session.as_ref().is_some_and(ended) {
            *session = None;
        }
    }

    /// Forgets the session if it is still the one with `session_public_key`,
    /// so a newer session started meanwhile stays in use.
    fn clear_key(&self, session_public_key: &str) {
        self.clear_if(|session| session.info.session_public_key == session_public_key);
    }
}

/// Sessions are dropped this many seconds before expiry so an in-flight
/// request never arrives at the server already expired.
const SESSION_EXPIRY_MARGIN_SECS: u64 = 30;

/// Which key signs a per-account request. Mirrors the server: requests are
/// wallet-signed unless the route is session-eligible.
#[derive(Clone, Copy, PartialEq, Eq)]
enum AuthMode {
    /// The delegated signer when a session is active, the wallet otherwise:
    /// reads, proposal list/get and proposal create/sign (issue #219).
    Session,
    /// Always the wallet.
    Wallet,
}

impl GuardianClient {
    /// Creates a new client connected to the specified GUARDIAN server endpoint.
    ///
    /// # Arguments
    /// * `endpoint` - The gRPC endpoint URL (e.g., "http://localhost:50051")
    pub async fn connect(endpoint: impl Into<String>) -> ClientResult<Self> {
        let endpoint = endpoint.into();
        let client = GuardianGrpcClient::connect(endpoint).await?;
        Ok(Self {
            client,
            auth: None,
            signer: None,
            session: SessionSlot::default(),
            last_timestamp: AtomicI64::new(0),
        })
    }

    /// Monotonic timestamp for auth metadata: strictly increasing across
    /// calls within a single client instance so concurrent or rapid-fire
    /// requests never mint duplicate `x-timestamp` values. Mirrors the TS
    /// client's `nextTimestamp`.
    fn next_timestamp(&self) -> i64 {
        next_monotonic_timestamp(&self.last_timestamp, Utc::now().timestamp_millis())
    }

    /// Configures scheme-aware authentication for authenticated GUARDIAN requests.
    pub fn with_auth(mut self, auth: Auth) -> Self {
        self.auth = Some(auth);
        self
    }

    /// Configures the signer used for authenticated GUARDIAN requests.
    pub fn with_signer(mut self, signer: Arc<dyn Signer>) -> Self {
        self.signer = Some(signer);
        self
    }

    /// Returns the hex-encoded public key of the configured auth or signer.
    pub fn auth_pubkey_hex(&self) -> Result<String, ClientError> {
        self.auth
            .as_ref()
            .map(|auth| auth.public_key_hex())
            .or_else(|| self.signer.as_ref().map(|signer| signer.public_key_hex()))
            .ok_or_else(|| {
                ClientError::InvalidResponse("GUARDIAN client has no signer configured".to_string())
            })
    }

    /// Returns the hex-encoded public key of the configured signer, if any.
    pub fn signer_pubkey_hex(&self) -> Result<String, ClientError> {
        self.auth_pubkey_hex()
    }

    /// Signs `request` and returns the session public key when the delegated
    /// signer signed it.
    fn add_auth_metadata(
        &self,
        request: &mut tonic::Request<impl prost::Message + std::fmt::Debug>,
        account_id: &AccountId,
        mode: AuthMode,
    ) -> ClientResult<Option<String>> {
        let request_payload = AuthRequestPayload::from_protobuf_message(request.get_ref());
        let timestamp = self.next_timestamp();

        if mode == AuthMode::Session {
            let digest =
                AuthRequestMessage::new(*account_id, timestamp, request_payload.clone()).to_word();
            if let Some((session_public_key, signature_hex)) = self.with_active_session(|session| {
                (
                    session.info.session_public_key.clone(),
                    session.key.sign_hex(digest),
                )
            }) {
                attach_auth_headers(request, &session_public_key, &signature_hex, timestamp)?;
                request
                    .metadata_mut()
                    .insert("x-auth-format", MetadataValue::from_static("session"));
                return Ok(Some(session_public_key));
            }
        }

        let (pubkey_hex, signature_hex) = if let Some(auth) = &self.auth {
            let pubkey_hex = auth.public_key_hex();
            let signature_hex = auth.sign_request_message(account_id, timestamp, request_payload);
            (pubkey_hex, signature_hex)
        } else if let Some(signer) = &self.signer {
            let pubkey_hex = signer.public_key_hex();
            let digest = AuthRequestMessage::new(*account_id, timestamp, request_payload).to_word();
            let signature_hex = signer.sign_word_hex(digest);
            (pubkey_hex, signature_hex)
        } else {
            return Ok(None);
        };

        attach_auth_headers(request, &pubkey_hex, &signature_hex, timestamp)?;
        Ok(None)
    }

    /// Attach lookup-bound auth metadata to a `GetAccountByKeyCommitment`
    /// request. Account-less because lookup is the operation that discovers
    /// the account ID, and signs the dedicated `LookupAuthMessage` digest —
    /// domain-separated from `AuthRequestMessage` by construction.
    ///
    /// The server derives the public key from the signature itself (Falcon
    /// embeds it; ECDSA recovers it). The `x-pubkey` header is sent for
    /// API consistency; raw lookup derives identity from the signature.
    fn add_lookup_auth_metadata<T: prost::Message + std::fmt::Debug>(
        &self,
        request: &mut tonic::Request<T>,
        key_commitment: Word,
    ) -> ClientResult<()> {
        let timestamp = self.next_timestamp();
        let digest = LookupAuthMessage::new(timestamp, key_commitment).to_word();

        let (pubkey_hex, signature_hex) = if let Some(auth) = &self.auth {
            (auth.public_key_hex(), auth.sign_word_hex(digest))
        } else if let Some(signer) = &self.signer {
            (signer.public_key_hex(), signer.sign_word_hex(digest))
        } else {
            return Err(ClientError::InvalidResponse(
                "GUARDIAN client has no signer configured".to_string(),
            ));
        };

        attach_auth_headers(request, &pubkey_hex, &signature_hex, timestamp)
    }

    /// Send an authenticated request, retrying only replay-classified
    /// rejections (`authentication_replay`) up to [`REPLAY_RETRY_LIMIT`]
    /// times. Every attempt rebuilds the request from the identical message
    /// with a fresh monotonic timestamp, recomputed digest, and fresh
    /// signature. Terminal failures (invalid signature, unauthorized signer,
    /// clock outside the skew window) propagate immediately. Signed by the
    /// wallet; session-eligible routes use [`Self::send_allowing_session`].
    async fn send_with_replay_retry<Req, Resp, F>(
        &mut self,
        account_id: &AccountId,
        message: Req,
        send: F,
    ) -> ClientResult<Resp>
    where
        Req: prost::Message + Clone + std::fmt::Debug,
        F: AsyncFnMut(
            &mut GuardianGrpcClient<Channel>,
            tonic::Request<Req>,
        ) -> Result<tonic::Response<Resp>, tonic::Status>,
    {
        self.send_with_replay_retry_as(account_id, message, AuthMode::Wallet, send)
            .await
    }

    /// [`Self::send_with_replay_retry`] for a session-eligible route: signed
    /// by the delegated signer while a session is active.
    async fn send_allowing_session<Req, Resp, F>(
        &mut self,
        account_id: &AccountId,
        message: Req,
        send: F,
    ) -> ClientResult<Resp>
    where
        Req: prost::Message + Clone + std::fmt::Debug,
        F: AsyncFnMut(
            &mut GuardianGrpcClient<Channel>,
            tonic::Request<Req>,
        ) -> Result<tonic::Response<Resp>, tonic::Status>,
    {
        self.send_with_replay_retry_as(account_id, message, AuthMode::Session, send)
            .await
    }

    /// Sends with `mode`, retrying `authentication_replay`. A session Guardian
    /// ended or no longer accepts (unknown key after a restart, rotated
    /// Guardian key or network) is dead for good, so it is dropped and the
    /// wallet signs from then on; the caller can start a new one.
    async fn send_with_replay_retry_as<Req, Resp, F>(
        &mut self,
        account_id: &AccountId,
        message: Req,
        mode: AuthMode,
        mut send: F,
    ) -> ClientResult<Resp>
    where
        Req: prost::Message + Clone + std::fmt::Debug,
        F: AsyncFnMut(
            &mut GuardianGrpcClient<Channel>,
            tonic::Request<Req>,
        ) -> Result<tonic::Response<Resp>, tonic::Status>,
    {
        self.discard_expired_session();
        let mut retries_left = REPLAY_RETRY_LIMIT;
        loop {
            let mut request = tonic::Request::new(message.clone());
            let session_key = self.add_auth_metadata(&mut request, account_id, mode)?;
            match send(&mut self.client, request).await {
                Ok(response) => return Ok(response.into_inner()),
                Err(status) => {
                    let error = ClientError::from(status);
                    if let Some(session_public_key) = &session_key
                        && error.is_rejected_session()
                    {
                        self.session.clear_key(session_public_key);
                    }
                    if retries_left == 0 || !error.is_replay_rejection() {
                        return Err(error);
                    }
                    retries_left -= 1;
                    tokio::time::sleep(REPLAY_RETRY_BACKOFF).await;
                }
            }
        }
    }

    /// Generates a delegated signer, has the configured wallet sign its grant,
    /// and registers it. Afterwards per-account reads and proposal requests
    /// are signed by the session key; `configure`, `push_delta`,
    /// `abandon_candidate`, account lookup and `revoke_all_sessions` keep
    /// using the wallet. A session already active is replaced, not revoked.
    /// The lifetime counts in whole seconds, and the grant names no website:
    /// a native client is not a browser page.
    pub async fn start_session(
        &mut self,
        options: StartSessionOptions,
    ) -> ClientResult<SessionInfo> {
        let ttl_secs = options.ttl.as_secs();
        if ttl_secs <= MIN_SESSION_TTL.as_secs() || ttl_secs > MAX_SESSION_TTL.as_secs() {
            return Err(ClientError::InvalidResponse(format!(
                "Session lifetime must be more than {}s and at most {}s",
                MIN_SESSION_TTL.as_secs(),
                MAX_SESSION_TTL.as_secs()
            )));
        }
        let scheme = self.wallet()?.scheme();
        let guardian_commitment = self.get_pubkey(Some(scheme.as_str())).await?.0;

        let issued_at = Utc::now().timestamp().max(0) as u64;
        let expires_at = issued_at + ttl_secs;
        let key = SessionKey::generate();
        let wallet = self.wallet()?;
        let commitment = wallet.commitment();
        let grant = SessionGrant::new(
            commitment,
            &key.public_key(),
            "",
            issued_at,
            expires_at,
            Word::from_hex(&guardian_commitment).map_err(|e| {
                ClientError::InvalidResponse(format!("Invalid Guardian commitment: {e}"))
            })?,
            options.network.as_str(),
        )
        .map_err(ClientError::InvalidResponse)?;
        let request = crate::proto::CreateSessionRequest {
            scheme: scheme.as_str().to_string(),
            auth_format: None,
            public_key: matches!(scheme, guardian_shared::SignatureScheme::Ecdsa)
                .then(|| wallet.public_key_hex()),
            signature: wallet.sign_word_hex(grant.to_word()),
            grant: Some(SessionGrantFields {
                signer_commitment: commitment.into_hex(),
                session_public_key: key.public_key_hex(),
                issued_at,
                expires_at,
                guardian_commitment,
                network: options.network,
                origin: String::new(),
            }),
        };
        let response = self
            .client
            .create_session(tonic::Request::new(request))
            .await?
            .into_inner();

        let info = SessionInfo {
            signer_commitment: response.signer_commitment,
            session_public_key: key.public_key_hex(),
            expires_at,
        };
        *self.session.lock() = Some(ActiveSession {
            key,
            info: info.clone(),
        });
        Ok(info)
    }

    /// The active session, if one is registered and not about to expire.
    pub fn session(&self) -> Option<SessionInfo> {
        self.with_active_session(|session| session.info.clone())
    }

    /// Shares `slot` as this client's session, so clients created for each
    /// operation use one session (see [`SessionSlot`]).
    pub fn with_session_slot(mut self, slot: SessionSlot) -> Self {
        self.session = slot;
        self
    }

    /// The session holder, to share with other clients.
    pub fn session_slot(&self) -> SessionSlot {
        self.session.clone()
    }

    /// Revokes the active session on the server and stops using it. Returns
    /// whether the server still had it active. On error the session stays in
    /// use, so the call can be retried.
    pub async fn revoke_session(&mut self) -> ClientResult<bool> {
        let timestamp = self.next_timestamp();
        let Some((session_public_key, signature_hex)) =
            self.session.lock().as_ref().map(|session| {
                let digest =
                    SessionLogoutMessage::new(&session.key.public_key(), timestamp).to_word();
                (
                    session.info.session_public_key.clone(),
                    session.key.sign_hex(digest),
                )
            })
        else {
            return Ok(false);
        };
        let mut request = tonic::Request::new(RevokeSessionRequest {});
        attach_auth_headers(&mut request, &session_public_key, &signature_hex, timestamp)?;
        let revoked = self
            .client
            .revoke_session(request)
            .await?
            .into_inner()
            .revoked;
        self.session.clear_key(&session_public_key);
        Ok(revoked)
    }

    /// Revokes every session of the configured wallet on this Guardian,
    /// including sessions started elsewhere, and stops using the session
    /// held when the call was made if that wallet granted it. A session a
    /// client sharing this one's [`SessionSlot`] starts while the call is in
    /// flight was issued after it and stays in use. Signed by the wallet.
    /// Returns how many sessions the server revoked.
    pub async fn revoke_all_sessions(&mut self) -> ClientResult<u64> {
        let timestamp = self.next_timestamp();
        let (commitment, pubkey_hex, signature_hex) = {
            let wallet = self.wallet()?;
            let commitment = wallet.commitment();
            let digest = SessionRevokeAllMessage::new(commitment, timestamp).to_word();
            (
                commitment,
                wallet.public_key_hex(),
                wallet.sign_word_hex(digest),
            )
        };
        let commitment = commitment.into_hex();
        let revoking = self.session.lock().as_ref().and_then(|session| {
            session
                .info
                .signer_commitment
                .eq_ignore_ascii_case(&commitment)
                .then(|| session.info.session_public_key.clone())
        });
        let mut request = tonic::Request::new(RevokeAllSessionsRequest {
            signer_commitment: commitment,
        });
        attach_auth_headers(&mut request, &pubkey_hex, &signature_hex, timestamp)?;
        let revoked = self
            .client
            .revoke_all_sessions(request)
            .await?
            .into_inner()
            .revoked;
        if let Some(session_public_key) = revoking {
            self.session.clear_key(&session_public_key);
        }
        Ok(revoked)
    }

    /// Forgets the session key once its grant is about to expire.
    fn discard_expired_session(&mut self) {
        let now = Utc::now().timestamp().max(0) as u64;
        self.session
            .clear_if(|session| now + SESSION_EXPIRY_MARGIN_SECS >= session.info.expires_at);
    }

    /// Runs `f` on the active session: registered, not about to expire, and
    /// granted by the wallet currently configured (a session never signs for
    /// another).
    fn with_active_session<T>(&self, f: impl FnOnce(&ActiveSession) -> T) -> Option<T> {
        let now = Utc::now().timestamp().max(0) as u64;
        let wallet = self.wallet().ok()?.commitment_hex();
        self.session
            .lock()
            .as_ref()
            .filter(|session| {
                now + SESSION_EXPIRY_MARGIN_SECS < session.info.expires_at
                    && session.info.signer_commitment.eq_ignore_ascii_case(&wallet)
            })
            .map(f)
    }

    /// The wallet: the configured auth, else the configured signer.
    fn wallet(&self) -> ClientResult<&dyn Signer> {
        self.auth
            .as_ref()
            .map(|auth| auth as &dyn Signer)
            .or(self.signer.as_deref())
            .ok_or_else(|| {
                ClientError::InvalidResponse("GUARDIAN client has no signer configured".to_string())
            })
    }

    /// Configure a new account
    ///
    /// # Arguments
    pub async fn configure(
        &mut self,
        account_id: &AccountId,
        auth: AuthConfig,
        initial_state: impl serde::Serialize,
    ) -> ClientResult<ConfigureResponse> {
        let initial_state_json = serde_json::to_string(&initial_state)?;

        let message = ConfigureRequest {
            account_id: account_id.to_string(),
            auth: Some(auth),
            initial_state: initial_state_json,
        };

        let inner = self
            .send_with_replay_retry(account_id, message, async |client, request| {
                client.configure(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Pushes a delta (state change) to the GUARDIAN server.
    ///
    /// This makes the delta canonical and triggers the canonicalization workflow.
    pub async fn push_delta(
        &mut self,
        account_id: &AccountId,
        nonce: u64,
        prev_commitment: impl Into<String>,
        delta_payload: impl serde::Serialize,
    ) -> ClientResult<PushDeltaResponse> {
        let delta_payload_json = serde_json::to_string(&delta_payload)?;

        let message = PushDeltaRequest {
            account_id: account_id.to_string(),
            nonce,
            prev_commitment: prev_commitment.into(),
            delta_payload: delta_payload_json,
        };

        let inner = self
            .send_with_replay_retry(account_id, message, async |client, request| {
                client.push_delta(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Retrieves a specific delta by nonce.
    pub async fn get_delta(
        &mut self,
        account_id: &AccountId,
        nonce: u64,
    ) -> ClientResult<GetDeltaResponse> {
        let message = GetDeltaRequest {
            account_id: account_id.to_string(),
            nonce,
        };

        let inner = self
            .send_allowing_session(account_id, message, async |client, request| {
                client.get_delta(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Retrieves all deltas starting from a given nonce.
    pub async fn get_delta_since(
        &mut self,
        account_id: &AccountId,
        from_nonce: u64,
    ) -> ClientResult<GetDeltaSinceResponse> {
        let message = GetDeltaSinceRequest {
            account_id: account_id.to_string(),
            from_nonce,
        };

        let inner = self
            .send_allowing_session(account_id, message, async |client, request| {
                client.get_delta_since(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Retrieves one page of the account's canonical delta history
    /// (issue #413), newest-first by nonce, with decoded input/output
    /// note summaries.
    ///
    /// `limit` is the page size in `[1, 500]` (server default 50 when
    /// `None`); `cursor` is the opaque `next_cursor` from a previous
    /// page (`None` for the first page). A `None` `next_cursor` on the
    /// response means the feed is exhausted.
    pub async fn get_delta_history(
        &mut self,
        account_id: &AccountId,
        limit: Option<u32>,
        cursor: Option<String>,
    ) -> ClientResult<GetDeltaHistoryResponse> {
        let message = GetDeltaHistoryRequest {
            account_id: account_id.to_string(),
            limit,
            cursor,
        };

        let inner = self
            .send_allowing_session(account_id, message, async |client, request| {
                client.get_delta_history(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Retrieves the current state for an account.
    pub async fn get_state(&mut self, account_id: &AccountId) -> ClientResult<GetStateResponse> {
        let message = GetStateRequest {
            account_id: account_id.to_string(),
        };

        let inner = self
            .send_allowing_session(account_id, message, async |client, request| {
                client.get_state(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Retrieves the nonce and commitment of the latest canonical state
    /// without the state blob. Mirror of HTTP `GET /state/nonce`.
    ///
    /// A client can skip [`Self::get_state`] when the returned nonce is below
    /// its local account nonce, or equal to it with the same commitment
    /// (issue #191). An equal nonce at a different commitment means the local
    /// account diverged from GUARDIAN, so it fetches the state.
    pub async fn get_canonical_nonce(
        &mut self,
        account_id: &AccountId,
    ) -> ClientResult<GetCanonicalNonceResponse> {
        let message = GetCanonicalNonceRequest {
            account_id: account_id.to_string(),
        };

        let inner = self
            .send_allowing_session(account_id, message, async |client, request| {
                client.get_canonical_nonce(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Look up the set of account IDs whose authorization set contains the
    /// given public-key commitment. Mirror of HTTP `GET /state/lookup`.
    ///
    /// Authentication is by proof-of-possession: the configured signer (or
    /// `Auth`) must hold the private key behind `key_commitment`. The server
    /// rejects any caller whose pubkey does not derive to the queried
    /// commitment, so this method MUST be called with a signer whose
    /// `commitment_hex` matches `key_commitment`.
    ///
    /// Returns an empty list if no account authorizes the commitment; the
    /// server intentionally does NOT distinguish that case from "wrong key"
    /// at the protocol level.
    pub async fn lookup_account_by_key_commitment(
        &mut self,
        key_commitment: &str,
    ) -> ClientResult<GetAccountByKeyCommitmentResponse> {
        let key_commitment_word = Word::from_hex(key_commitment).map_err(|e| {
            ClientError::InvalidResponse(format!("Invalid key_commitment hex: {e}"))
        })?;

        let mut request = tonic::Request::new(GetAccountByKeyCommitmentRequest {
            key_commitment: key_commitment.to_string(),
        });

        self.add_lookup_auth_metadata(&mut request, key_commitment_word)?;

        let response = self.client.get_account_by_key_commitment(request).await?;
        Ok(response.into_inner())
    }

    /// Retrieves the GUARDIAN server's public key commitment (and optionally the raw public key).
    pub async fn get_pubkey(
        &mut self,
        scheme: Option<&str>,
    ) -> ClientResult<(String, Option<String>)> {
        let request = tonic::Request::new(GetPubkeyRequest {
            scheme: scheme.map(|s| s.to_string()),
        });
        let response = self.client.get_pubkey(request).await?;
        let inner = response.into_inner();
        Ok((inner.pubkey, inner.raw_pubkey))
    }

    /// Push a delta proposal
    pub async fn push_delta_proposal(
        &mut self,
        account_id: &AccountId,
        nonce: u64,
        delta_payload: impl serde::Serialize,
    ) -> ClientResult<PushDeltaProposalResponse> {
        let delta_payload_json = serde_json::to_string(&delta_payload)?;

        let message = PushDeltaProposalRequest {
            account_id: account_id.to_string(),
            nonce,
            delta_payload: delta_payload_json,
        };

        let inner = self
            .send_allowing_session(account_id, message, async |client, request| {
                client.push_delta_proposal(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Get all delta proposals for an account
    pub async fn get_delta_proposals(
        &mut self,
        account_id: &AccountId,
    ) -> ClientResult<GetDeltaProposalsResponse> {
        let message = GetDeltaProposalsRequest {
            account_id: account_id.to_string(),
        };

        let inner = self
            .send_allowing_session(account_id, message, async |client, request| {
                client.get_delta_proposals(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Get a specific delta proposal for an account by commitment.
    pub async fn get_delta_proposal(
        &mut self,
        account_id: &AccountId,
        commitment: impl Into<String>,
    ) -> ClientResult<GetDeltaProposalResponse> {
        let message = GetDeltaProposalRequest {
            account_id: account_id.to_string(),
            commitment: commitment.into(),
        };

        let inner = self
            .send_allowing_session(account_id, message, async |client, request| {
                client.get_delta_proposal(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Sign a delta proposal
    pub async fn sign_delta_proposal(
        &mut self,
        account_id: &AccountId,
        commitment: impl Into<String>,
        signature: JsonProposalSignature,
    ) -> ClientResult<SignDeltaProposalResponse> {
        let proto_signature = Some(proto_signature_from_json(&signature));

        let message = SignDeltaProposalRequest {
            account_id: account_id.to_string(),
            commitment: commitment.into(),
            signature: proto_signature,
        };

        let inner = self
            .send_allowing_session(account_id, message, async |client, request| {
                client.sign_delta_proposal(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }

    /// Request abandonment of a pending canonicalization candidate whose
    /// transaction will never land on-chain (issue #319).
    ///
    /// The request records an abandon *intent*: the delta stays a
    /// candidate — the account stays locked — until the guardian's
    /// canonicalization worker confirms over the abandon quarantine that
    /// the transaction did not land, then discards the delta as
    /// `client_abandoned` and releases the account (typically well under
    /// a minute). Poll `get_delta` for the resolution; the returned
    /// `state` is `"pending"` or, when already resolved, `"abandoned"`.
    ///
    /// `nonce` pins the exact candidate (learned from the delta feed).
    /// The server refuses with `GUARDIAN_CANDIDATE_LANDED` when the
    /// transaction actually landed. Retries are idempotent and preserve
    /// the original request timestamp.
    pub async fn abandon_candidate(
        &mut self,
        account_id: &AccountId,
        nonce: u64,
    ) -> ClientResult<AbandonDeltaCandidateResponse> {
        let message = AbandonDeltaCandidateRequest {
            account_id: account_id.to_string(),
            nonce,
        };

        let inner = self
            .send_with_replay_retry(account_id, message, async |client, request| {
                client.abandon_delta_candidate(request).await
            })
            .await?;

        if !inner.success {
            return Err(ClientError::ServerError(inner.message.clone()));
        }

        Ok(inner)
    }
}

/// `max(now_ms, last + 1)`: the wall clock when it is ahead, otherwise one
/// past the previously minted value. Shared by all auth paths of one client
/// instance; the caller supplies the clock so the arithmetic stays
/// deterministic under test.
fn next_monotonic_timestamp(last_timestamp: &AtomicI64, now_ms: i64) -> i64 {
    let mut previous = last_timestamp.load(Ordering::SeqCst);
    loop {
        let next = previous.saturating_add(1).max(now_ms);
        match last_timestamp.compare_exchange_weak(
            previous,
            next,
            Ordering::SeqCst,
            Ordering::SeqCst,
        ) {
            Ok(_) => return next,
            Err(actual) => previous = actual,
        }
    }
}

fn attach_auth_headers<T: prost::Message>(
    request: &mut tonic::Request<T>,
    pubkey_hex: &str,
    signature_hex: &str,
    timestamp: i64,
) -> ClientResult<()> {
    let pubkey_metadata = MetadataValue::try_from(pubkey_hex)
        .map_err(|e| ClientError::InvalidResponse(format!("Invalid pubkey: {e}")))?;
    let signature_metadata = MetadataValue::try_from(signature_hex)
        .map_err(|e| ClientError::InvalidResponse(format!("Invalid signature: {e}")))?;
    let timestamp_metadata = MetadataValue::try_from(timestamp.to_string())
        .map_err(|e| ClientError::InvalidResponse(format!("Invalid timestamp: {e}")))?;

    let metadata = request.metadata_mut();
    metadata.insert("x-pubkey", pubkey_metadata);
    metadata.insert("x-signature", signature_metadata);
    metadata.insert("x-timestamp", timestamp_metadata);
    Ok(())
}

fn proto_signature_from_json(signature: &JsonProposalSignature) -> ProtoProposalSignature {
    match signature {
        JsonProposalSignature::Falcon { signature } => ProtoProposalSignature {
            scheme: "falcon".to_string(),
            signature: signature.clone(),
            public_key: None,
            message_format: String::new(),
        },
        JsonProposalSignature::Ecdsa {
            signature,
            public_key,
            message_format,
        } => ProtoProposalSignature {
            scheme: "ecdsa".to_string(),
            signature: signature.clone(),
            public_key: public_key.clone(),
            message_format: match message_format {
                guardian_shared::EcdsaMessageFormat::Raw => String::new(),
                guardian_shared::EcdsaMessageFormat::Eip712 => "eip712".to_string(),
            },
        },
    }
}

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

    #[test]
    fn monotonic_timestamp_follows_the_clock_when_ahead() {
        let last = AtomicI64::new(0);
        assert_eq!(next_monotonic_timestamp(&last, 1_000), 1_000);
        assert_eq!(next_monotonic_timestamp(&last, 2_000), 2_000);
    }

    #[test]
    fn monotonic_timestamp_never_repeats_within_one_instance() {
        let last = AtomicI64::new(0);
        let first = next_monotonic_timestamp(&last, 1_000);
        let second = next_monotonic_timestamp(&last, 1_000);
        let third = next_monotonic_timestamp(&last, 1_000);
        assert_eq!((first, second, third), (1_000, 1_001, 1_002));
    }

    #[test]
    fn monotonic_timestamp_survives_a_clock_step_backwards() {
        let last = AtomicI64::new(0);
        assert_eq!(next_monotonic_timestamp(&last, 5_000), 5_000);
        assert_eq!(next_monotonic_timestamp(&last, 4_000), 5_001);
    }

    #[test]
    fn concurrent_timestamps_are_unique() {
        let last = Arc::new(AtomicI64::new(0));
        let handles: Vec<_> = (0..8)
            .map(|_| {
                let last = last.clone();
                std::thread::spawn(move || {
                    (0..100)
                        .map(|_| next_monotonic_timestamp(&last, 1_000))
                        .collect::<Vec<_>>()
                })
            })
            .collect();
        let mut minted: Vec<i64> = handles
            .into_iter()
            .flat_map(|handle| handle.join().unwrap())
            .collect();
        let count = minted.len();
        minted.sort_unstable();
        minted.dedup();
        assert_eq!(
            minted.len(),
            count,
            "timestamps must be unique across threads"
        );
    }
}