katzenpost_thin_client 0.0.17

// SPDX-FileCopyrightText: Copyright (C) 2026 David Stainton
// SPDX-License-Identifier: AGPL-3.0-only

//! Pigeonhole protocol API for the thin client.
//!
//! This module provides methods for interacting with the Pigeonhole protocol,
//! including key generation, encryption, and ARQ (Automatic Repeat Request)
//! for reliable message delivery to the courier.

use std::collections::BTreeMap;
use serde_cbor::Value;
use log::debug;
use blake2::{Blake2b, Digest, digest::consts::U32};

use crate::error::{ThinClientError, error_code_to_error};
use crate::core::ThinClient;

// ========================================================================
// Helper module for serializing Option<Vec<u8>> as CBOR byte strings
// ========================================================================

mod optional_bytes {
    use serde::{Deserialize, Deserializer, Serialize, Serializer};

    pub fn serialize<S>(value: &Option<Vec<u8>>, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        match value {
            Some(bytes) => serde_bytes::serialize(bytes, serializer),
            None => Option::<&[u8]>::None.serialize(serializer),
        }
    }

    pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<Vec<u8>>, D::Error>
    where
        D: Deserializer<'de>,
    {
        let opt: Option<serde_bytes::ByteBuf> = Option::deserialize(deserializer)?;
        Ok(opt.map(|b| b.into_vec()))
    }
}

// ========================================================================
// NEW Pigeonhole API Protocol Message Structs
// ========================================================================

/// Request to create a new keypair for the Pigeonhole protocol.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct NewKeypairRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    seed: Vec<u8>,
}

/// Reply containing the generated keypair and first message index.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct NewKeypairReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(default, with = "optional_bytes")]
    write_cap: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    read_cap: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    first_message_index: Option<Vec<u8>>,
    #[serde(default)]
    error_code: u8,
}

/// Request to encrypt a read operation.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct EncryptReadRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    read_cap: Vec<u8>,
    #[serde(with = "serde_bytes")]
    message_box_index: Vec<u8>,
}

/// Reply containing the encrypted read operation.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct EncryptReadReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(default, with = "optional_bytes")]
    message_ciphertext: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    envelope_descriptor: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    envelope_hash: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    next_message_box_index: Option<Vec<u8>>,
    #[serde(default)]
    error_code: u8,
}

/// Request to encrypt a write operation.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct EncryptWriteRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    plaintext: Vec<u8>,
    #[serde(with = "serde_bytes")]
    write_cap: Vec<u8>,
    #[serde(with = "serde_bytes")]
    message_box_index: Vec<u8>,
}

/// Reply containing the encrypted write operation.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct EncryptWriteReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(default, with = "optional_bytes")]
    message_ciphertext: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    envelope_descriptor: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    envelope_hash: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    next_message_box_index: Option<Vec<u8>>,
    #[serde(default)]
    error_code: u8,
}

/// Request to start resending an encrypted message via ARQ.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct StartResendingEncryptedMessageRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(skip_serializing_if = "Option::is_none", with = "optional_bytes")]
    read_cap: Option<Vec<u8>>,
    #[serde(skip_serializing_if = "Option::is_none", with = "optional_bytes")]
    write_cap: Option<Vec<u8>>,
    #[serde(skip_serializing_if = "Option::is_none", with = "optional_bytes")]
    message_box_index: Option<Vec<u8>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    reply_index: Option<u8>,
    #[serde(with = "serde_bytes")]
    envelope_descriptor: Vec<u8>,
    #[serde(with = "serde_bytes")]
    message_ciphertext: Vec<u8>,
    #[serde(with = "serde_bytes")]
    envelope_hash: Vec<u8>,
    /// If true, BoxIDNotFound errors on reads trigger immediate error instead of automatic retries.
    #[serde(skip_serializing_if = "std::ops::Not::not")]
    no_retry_on_box_id_not_found: bool,
    /// If true, BoxAlreadyExists errors on writes are returned as errors instead of idempotent success.
    #[serde(skip_serializing_if = "std::ops::Not::not")]
    no_idempotent_box_already_exists: bool,
}

/// Reply containing the plaintext from a resent encrypted message.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct StartResendingEncryptedMessageReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(default, with = "optional_bytes")]
    plaintext: Option<Vec<u8>>,
    error_code: u8,
    #[serde(default, with = "optional_bytes")]
    courier_identity_hash: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    courier_queue_id: Option<Vec<u8>>,
}

/// Result returned by `start_resending_encrypted_message` and its variants.
///
/// Contains the decrypted plaintext (for reads) along with the courier identity
/// that was selected to handle the message. Callers can watch PKI document updates
/// and cancel+re-encrypt if the courier disappears from consensus.
#[derive(Debug, Clone)]
pub struct StartResendingResult {
    pub plaintext: Vec<u8>,
    pub courier_identity_hash: Option<Vec<u8>>,
    pub courier_queue_id: Option<Vec<u8>>,
}

/// Request to write a whole multi-box payload via the windowed SACK ARQ.
#[derive(Debug, Clone, serde::Serialize)]
struct WriteStreamRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    write_cap: Vec<u8>,
    #[serde(with = "serde_bytes")]
    start_index: Vec<u8>,
    #[serde(with = "serde_bytes")]
    payload: Vec<u8>,
    window: i64,
}

/// Reply to a WriteStream request.
#[derive(Debug, Clone, serde::Deserialize)]
struct WriteStreamReply {
    #[serde(default, with = "optional_bytes")]
    next_message_box_index: Option<Vec<u8>>,
    #[serde(default)]
    error_code: u8,
    #[serde(default)]
    box_count: u32,
}

/// Request to read many sequential boxes via the windowed SACK ARQ.
#[derive(Debug, Clone, serde::Serialize)]
struct ReadStreamRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    read_cap: Vec<u8>,
    #[serde(with = "serde_bytes")]
    start_index: Vec<u8>,
    box_count: u32,
    window: i64,
}

/// Reply to a ReadStream request.
#[derive(Debug, Clone, serde::Deserialize)]
struct ReadStreamReply {
    #[serde(default, with = "optional_bytes")]
    payload: Option<Vec<u8>>,
    #[serde(default, with = "optional_bytes")]
    next_message_box_index: Option<Vec<u8>>,
    #[serde(default)]
    error_code: u8,
    #[serde(default)]
    box_count: u32,
}

/// Request to cancel resending an encrypted message.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CancelResendingEncryptedMessageRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    envelope_hash: Vec<u8>,
}

/// Reply confirming cancellation of resending.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CancelResendingEncryptedMessageReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    error_code: u8,
}

/// Request to increment a MessageBoxIndex.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct NextMessageBoxIndexRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    message_box_index: Vec<u8>,
}

/// Reply containing the incremented MessageBoxIndex.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct NextMessageBoxIndexReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(default, with = "optional_bytes")]
    next_message_box_index: Option<Vec<u8>>,
    #[serde(default)]
    error_code: u8,
}

/// Request to read the BACAP Idx64 counter from a MessageBoxIndex.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct GetMessageBoxIndexCounterRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    message_box_index: Vec<u8>,
}

/// Reply containing the BACAP Idx64 counter.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct GetMessageBoxIndexCounterReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(default)]
    counter: u64,
    #[serde(default)]
    error_code: u8,
}

/// Request to start resending a copy command.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct StartResendingCopyCommandRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    write_cap: Vec<u8>,
    #[serde(skip_serializing_if = "Option::is_none", default, with = "optional_bytes")]
    courier_identity_hash: Option<Vec<u8>>,
    #[serde(skip_serializing_if = "Option::is_none", default, with = "optional_bytes")]
    courier_queue_id: Option<Vec<u8>>,
}

/// Reply confirming start of copy command resending.
///
/// `replica_error_code` and `failed_envelope_index` are populated only when
/// `error_code == 26` (`ThinClientErrorCopyCommandFailed`) — they carry
/// diagnostic detail about which replica rejected which envelope in the copy
/// stream. Absent tags decode to zero via `#[serde(default)]`, preserving
/// compatibility with daemons that omit them.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct StartResendingCopyCommandReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    error_code: u8,
    #[serde(default)]
    replica_error_code: u8,
    #[serde(default)]
    failed_envelope_index: u64,
}

/// Thin-client error code reserved for courier-reported copy command failures.
/// Mirrors `ThinClientErrorCopyCommandFailed` in the Go thin client.
const THIN_CLIENT_ERROR_COPY_COMMAND_FAILED: u8 = 26;

/// Maps a `StartResendingCopyCommandReply` to an error variant (or `None` on
/// success). Unlike `error_code_to_error()`, this helper has access to the
/// reply's diagnostic fields and can populate
/// `ThinClientError::CopyCommandFailed { .. }` when the courier reports code 26.
fn copy_reply_to_error(reply: &StartResendingCopyCommandReply) -> Option<ThinClientError> {
    match reply.error_code {
        0 => None,
        THIN_CLIENT_ERROR_COPY_COMMAND_FAILED => Some(ThinClientError::CopyCommandFailed {
            replica_error_code: reply.replica_error_code,
            failed_envelope_index: reply.failed_envelope_index,
        }),
        code => Some(error_code_to_error(code)),
    }
}

/// Request to cancel resending a copy command.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CancelResendingCopyCommandRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    write_cap_hash: Vec<u8>,
}

/// Reply confirming cancellation of copy command resending.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CancelResendingCopyCommandReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    error_code: u8,
}

/// Request to create courier envelopes from a payload.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CreateCourierEnvelopesFromPayloadRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    payload: Vec<u8>,
    #[serde(with = "serde_bytes")]
    dest_write_cap: Vec<u8>,
    #[serde(with = "serde_bytes")]
    dest_start_index: Vec<u8>,
    is_start: bool,
    is_last: bool,
}

/// Reply containing the created courier envelopes and next destination index.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CreateCourierEnvelopesFromPayloadReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    /// Envelopes is None when the daemon returns an error.
    envelopes: Option<Vec<serde_bytes::ByteBuf>>,
    /// Next destination message box index after all boxes consumed by this call.
    #[serde(default, with = "optional_bytes")]
    next_dest_index: Option<Vec<u8>>,
    #[serde(default)]
    error_code: u8,
}

/// A destination for creating courier envelopes.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct EnvelopeDestination {
    #[serde(with = "serde_bytes")]
    payload: Vec<u8>,
    #[serde(with = "serde_bytes")]
    write_cap: Vec<u8>,
    #[serde(with = "serde_bytes")]
    start_index: Vec<u8>,
}

/// Request to create courier envelopes from multiple payloads.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CreateCourierEnvelopesFromPayloadsRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    destinations: Vec<EnvelopeDestination>,
    is_start: bool,
    is_last: bool,
    #[serde(default, with = "optional_bytes")]
    buffer: Option<Vec<u8>>,
}

/// Reply containing the created courier envelopes from multiple payloads and buffer state.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CreateCourierEnvelopesFromPayloadsReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    /// Envelopes is None when the daemon returns an error.
    envelopes: Option<Vec<serde_bytes::ByteBuf>>,
    /// Buffer contains any data buffered by the encoder that hasn't been output yet.
    /// None when the daemon returns an error.
    #[serde(default, with = "optional_bytes")]
    buffer: Option<Vec<u8>>,
    /// Next destination indices for each destination, in request order.
    #[serde(default)]
    next_dest_indices: Option<Vec<serde_bytes::ByteBuf>>,
    #[serde(default)]
    error_code: u8,
}

/// Request to create courier envelopes from a range of tombstones.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CreateCourierEnvelopesFromTombstoneRangeRequest {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    #[serde(with = "serde_bytes")]
    dest_write_cap: Vec<u8>,
    #[serde(with = "serde_bytes")]
    dest_start_index: Vec<u8>,
    max_count: u32,
    is_start: bool,
    is_last: bool,
    #[serde(default, with = "optional_bytes")]
    buffer: Option<Vec<u8>>,
}

/// Reply containing the created tombstone courier envelopes.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CreateCourierEnvelopesFromTombstoneRangeReply {
    #[serde(with = "serde_bytes")]
    query_id: Vec<u8>,
    /// Envelopes is None when the daemon returns an error.
    envelopes: Option<Vec<serde_bytes::ByteBuf>>,
    /// Buffer contains any residual data for the next call.
    #[serde(default, with = "optional_bytes")]
    buffer: Option<Vec<u8>>,
    /// Next destination index after all tombstones created.
    #[serde(default, with = "optional_bytes")]
    next_dest_index: Option<Vec<u8>>,
    #[serde(default)]
    error_code: u8,
}

/// Result from new_keypair containing the generated capabilities.
#[derive(Debug, Clone)]
pub struct KeypairResult {
    pub write_cap: Vec<u8>,
    pub read_cap: Vec<u8>,
    pub first_message_index: Vec<u8>,
}

/// Result from encrypt_read containing the encrypted read request.
#[derive(Debug, Clone)]
pub struct EncryptReadResult {
    pub message_ciphertext: Vec<u8>,
    pub envelope_descriptor: Vec<u8>,
    pub envelope_hash: [u8; 32],
    pub next_message_box_index: Vec<u8>,
}

/// Result from encrypt_write containing the encrypted write request.
#[derive(Debug, Clone)]
pub struct EncryptWriteResult {
    pub message_ciphertext: Vec<u8>,
    pub envelope_descriptor: Vec<u8>,
    pub envelope_hash: [u8; 32],
    pub next_message_box_index: Vec<u8>,
}

/// Result of creating courier envelopes.
#[derive(Debug, Clone)]
pub struct CreateEnvelopesResult {
    /// The serialized CopyStreamElements to send to the network.
    pub envelopes: Vec<Vec<u8>>,
    /// The buffered data that hasn't been output yet. Persist this for crash recovery.
    /// Only populated by create_courier_envelopes_from_multi_payload.
    pub buffer: Vec<u8>,
    /// The next destination message box index after all boxes consumed by this call.
    /// Only populated by create_courier_envelopes_from_payload.
    pub next_dest_index: Option<Vec<u8>>,
    /// The next destination indices for each destination, in request order.
    /// Only populated by create_courier_envelopes_from_multi_payload.
    pub next_dest_indices: Option<Vec<Vec<u8>>>,
}

// ========================================================================
// NEW Pigeonhole API Methods
// ========================================================================

impl ThinClient {
    /// Creates a new keypair for use with the Pigeonhole protocol.
    ///
    /// This method generates a WriteCap and ReadCap from the provided seed using
    /// the BACAP (Blinding-and-Capability) protocol. The WriteCap should be stored
    /// securely for writing messages, while the ReadCap can be shared with others
    /// to allow them to read messages.
    ///
    /// # Arguments
    /// * `seed` - 32-byte seed used to derive the keypair
    ///
    /// # Returns
    /// * `Ok((write_cap, read_cap, first_message_index))` on success
    /// * `Err(ThinClientError)` on failure
    pub async fn new_keypair(&self, seed: &[u8; 32]) -> Result<KeypairResult, ThinClientError> {
        let query_id = Self::new_query_id();

        let request_inner = NewKeypairRequest {
            query_id: query_id.clone(),
            seed: seed.to_vec(),
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("new_keypair".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: NewKeypairReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("new_keypair failed with error code: {}", reply.error_code)));
        }

        let write_cap = reply.write_cap.ok_or_else(|| ThinClientError::Other("new_keypair: write_cap is None".to_string()))?;
        let read_cap = reply.read_cap.ok_or_else(|| ThinClientError::Other("new_keypair: read_cap is None".to_string()))?;
        let first_message_index = reply.first_message_index.ok_or_else(|| ThinClientError::Other("new_keypair: first_message_index is None".to_string()))?;

        Ok(KeypairResult { write_cap, read_cap, first_message_index })
    }

    /// Encrypts a read operation for a given read capability.
    ///
    /// This method prepares an encrypted read request that can be sent to the
    /// courier service to retrieve a message from a pigeonhole box.
    ///
    /// # Arguments
    /// * `read_cap` - Read capability that grants access to the channel
    /// * `message_box_index` - Starting read position for the channel
    ///
    /// # Returns
    /// * `Ok(EncryptReadResult)` on success
    /// * `Err(ThinClientError)` on failure
    pub async fn encrypt_read(
        &self,
        read_cap: &[u8],
        message_box_index: &[u8]
    ) -> Result<EncryptReadResult, ThinClientError> {
        let query_id = Self::new_query_id();

        let request_inner = EncryptReadRequest {
            query_id: query_id.clone(),
            read_cap: read_cap.to_vec(),
            message_box_index: message_box_index.to_vec(),
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("encrypt_read".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: EncryptReadReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("encrypt_read failed with error code: {}", reply.error_code)));
        }

        let message_ciphertext = reply.message_ciphertext.ok_or_else(|| ThinClientError::Other("encrypt_read: message_ciphertext is None".to_string()))?;
        let envelope_descriptor = reply.envelope_descriptor.ok_or_else(|| ThinClientError::Other("encrypt_read: envelope_descriptor is None".to_string()))?;
        let envelope_hash_vec = reply.envelope_hash.ok_or_else(|| ThinClientError::Other("encrypt_read: envelope_hash is None".to_string()))?;
        let next_message_box_index = reply.next_message_box_index.ok_or_else(|| ThinClientError::Other("encrypt_read: next_message_box_index is None".to_string()))?;

        let mut envelope_hash = [0u8; 32];
        envelope_hash.copy_from_slice(&envelope_hash_vec[..32]);

        Ok(EncryptReadResult {
            message_ciphertext,
            envelope_descriptor,
            envelope_hash,
            next_message_box_index,
        })
    }

    /// Encrypts a write operation for a given write capability.
    ///
    /// This method prepares an encrypted write request that can be sent to the
    /// courier service to store a message in a pigeonhole box.
    ///
    /// # Plaintext Size Constraint
    ///
    /// The `plaintext` must not exceed `PigeonholeGeometry.max_plaintext_payload_length` bytes.
    /// The daemon internally adds a 4-byte big-endian length prefix before padding and
    /// encryption, so the actual wire format is `[4-byte length][plaintext][zero padding]`.
    ///
    /// If the plaintext exceeds the maximum size, the daemon will return
    /// `ThinClientErrorInvalidRequest`.
    ///
    /// # Arguments
    /// * `plaintext` - The plaintext message to encrypt. Must be at most
    ///   `PigeonholeGeometry.max_plaintext_payload_length` bytes.
    /// * `write_cap` - Write capability that grants access to the channel.
    /// * `message_box_index` - The message box index for this write operation.
    ///
    /// # Returns
    /// * `Ok((message_ciphertext, envelope_descriptor, envelope_hash))` on success
    /// * `Err(ThinClientError)` on failure (including if plaintext is too large)
    pub async fn encrypt_write(
        &self,
        plaintext: &[u8],
        write_cap: &[u8],
        message_box_index: &[u8]
    ) -> Result<EncryptWriteResult, ThinClientError> {
        let query_id = Self::new_query_id();

        let request_inner = EncryptWriteRequest {
            query_id: query_id.clone(),
            plaintext: plaintext.to_vec(),
            write_cap: write_cap.to_vec(),
            message_box_index: message_box_index.to_vec(),
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("encrypt_write".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: EncryptWriteReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("encrypt_write failed with error code: {}", reply.error_code)));
        }

        let message_ciphertext = reply.message_ciphertext.ok_or_else(|| ThinClientError::Other("encrypt_write: message_ciphertext is None".to_string()))?;
        let envelope_descriptor = reply.envelope_descriptor.ok_or_else(|| ThinClientError::Other("encrypt_write: envelope_descriptor is None".to_string()))?;
        let envelope_hash_vec = reply.envelope_hash.ok_or_else(|| ThinClientError::Other("encrypt_write: envelope_hash is None".to_string()))?;
        let next_message_box_index = reply.next_message_box_index.ok_or_else(|| ThinClientError::Other("encrypt_write: next_message_box_index is None".to_string()))?;

        let mut envelope_hash = [0u8; 32];
        envelope_hash.copy_from_slice(&envelope_hash_vec[..32]);

        Ok(EncryptWriteResult {
            message_ciphertext,
            envelope_descriptor,
            envelope_hash,
            next_message_box_index,
        })
    }

    /// Sends an encrypted message via ARQ (Automatic Repeat Request) and blocks until completion.
    ///
    /// This method BLOCKS until a reply is received from the daemon.
    /// The message will be resent periodically until either:
    /// - A successful response is received
    /// - An error response is received from the daemon
    /// - The operation is cancelled via cancel_resending_encrypted_message
    ///
    /// The daemon implements a finite state machine for the stop-and-wait ARQ protocol:
    /// - **Default writes** (write_cap set, no_idempotent_box_already_exists false):
    ///   Returns success after a single round-trip. The courier ACK confirms the
    ///   envelope was received and will be dispatched to both shard replicas.
    /// - **BoxAlreadyExists-aware writes** (no_idempotent_box_already_exists true):
    ///   Requires two round-trips — one for the courier ACK, and a second to
    ///   retrieve the replica's error code.
    /// - **Reads** (read_cap set): Requires two round-trips — one for the courier
    ///   ACK, and a second to retrieve the decrypted payload from the replica.
    ///
    /// # Arguments
    /// * `read_cap` - Optional read capability (for read operations)
    /// * `write_cap` - Optional write capability (for write operations)
    /// * `message_box_index` - Current message box index being operated on (for read operations)
    /// * `reply_index` - Reply index for the operation (None for tombstone writes)
    /// * `envelope_descriptor` - Envelope descriptor from encrypt_read/encrypt_write
    /// * `message_ciphertext` - Encrypted message from encrypt_read/encrypt_write
    /// * `envelope_hash` - Envelope hash from encrypt_read/encrypt_write
    ///
    /// # Returns
    /// * `Ok(plaintext)` - For read operations, the decrypted plaintext message
    ///   (at most `PigeonholeGeometry.max_plaintext_payload_length` bytes).
    ///   For write operations, returns an empty vector on success.
    /// * `Err(ThinClientError)` on failure
    pub async fn start_resending_encrypted_message(
        &self,
        read_cap: Option<&[u8]>,
        write_cap: Option<&[u8]>,
        message_box_index: Option<&[u8]>,
        reply_index: Option<u8>,
        envelope_descriptor: &[u8],
        message_ciphertext: &[u8],
        envelope_hash: &[u8; 32]
    ) -> Result<StartResendingResult, ThinClientError> {
        self.start_resending_encrypted_message_with_options(
            read_cap,
            write_cap,
            message_box_index,
            reply_index,
            envelope_descriptor,
            message_ciphertext,
            envelope_hash,
            false,
            false,
        ).await
    }

    /// Behaves exactly like `start_resending_encrypted_message` save that
    /// it returns `ThinClientError::BoxAlreadyExists` when the replica
    /// reports the destination box has already been written, rather than
    /// swallowing the condition as idempotent success. Use this when one
    /// needs to distinguish a fresh write from a repeat: for instance,
    /// when implementing optimistic concurrency on top of the channel, or
    /// when establishing whether a particular call actually caused a state
    /// change at the replica.
    ///
    /// Note that this variant costs an additional mixnet round trip: the
    /// BoxAlreadyExists code is carried by the replica's reply rather than
    /// the courier's ACK, so the daemon must dispatch a second SURB before
    /// it can return the answer.
    ///
    /// As with `start_resending_encrypted_message`, an in-flight call may
    /// be cancelled from another task via
    /// `cancel_resending_encrypted_message`.
    ///
    /// # Arguments
    /// Same as `start_resending_encrypted_message`
    ///
    /// # Returns
    /// * `Ok(plaintext)` on success
    /// * `Err(ThinClientError::BoxAlreadyExists)` if the box already contains data
    /// * `Err(ThinClientError)` on other failures
    pub async fn start_resending_encrypted_message_return_box_exists(
        &self,
        read_cap: Option<&[u8]>,
        write_cap: Option<&[u8]>,
        message_box_index: Option<&[u8]>,
        reply_index: Option<u8>,
        envelope_descriptor: &[u8],
        message_ciphertext: &[u8],
        envelope_hash: &[u8; 32]
    ) -> Result<StartResendingResult, ThinClientError> {
        self.start_resending_encrypted_message_with_options(
            read_cap,
            write_cap,
            message_box_index,
            reply_index,
            envelope_descriptor,
            message_ciphertext,
            envelope_hash,
            false,
            true,  // no_idempotent_box_already_exists
        ).await
    }

    /// Behaves exactly like `start_resending_encrypted_message` save that
    /// it disables the daemon's automatic retry of
    /// `ThinClientError::BoxIdNotFound`. The caller learns at once that
    /// the box is absent rather than waiting for replication to settle.
    ///
    /// Use this when polling a box that may not yet have been written:
    /// for instance, when a reader peeks ahead at a peer's next message
    /// before that peer has produced it. The regular variant would block
    /// until the box appeared, which can be many round trips.
    ///
    /// As with `start_resending_encrypted_message`, an in-flight call may
    /// be cancelled from another task via
    /// `cancel_resending_encrypted_message`.
    ///
    /// # Arguments
    /// Same as `start_resending_encrypted_message`
    ///
    /// # Returns
    /// * `Ok(plaintext)` on success
    /// * `Err(ThinClientError::BoxIdNotFound)` if the box does not exist (no automatic retries)
    /// * `Err(ThinClientError)` on other failures
    pub async fn start_resending_encrypted_message_no_retry(
        &self,
        read_cap: Option<&[u8]>,
        write_cap: Option<&[u8]>,
        message_box_index: Option<&[u8]>,
        reply_index: Option<u8>,
        envelope_descriptor: &[u8],
        message_ciphertext: &[u8],
        envelope_hash: &[u8; 32]
    ) -> Result<StartResendingResult, ThinClientError> {
        self.start_resending_encrypted_message_with_options(
            read_cap,
            write_cap,
            message_box_index,
            reply_index,
            envelope_descriptor,
            message_ciphertext,
            envelope_hash,
            true,  // no_retry_on_box_id_not_found
            false,
        ).await
    }

    /// Internal method with all options for start_resending_encrypted_message.
    async fn start_resending_encrypted_message_with_options(
        &self,
        read_cap: Option<&[u8]>,
        write_cap: Option<&[u8]>,
        message_box_index: Option<&[u8]>,
        reply_index: Option<u8>,
        envelope_descriptor: &[u8],
        message_ciphertext: &[u8],
        envelope_hash: &[u8; 32],
        no_retry_on_box_id_not_found: bool,
        no_idempotent_box_already_exists: bool,
    ) -> Result<StartResendingResult, ThinClientError> {
        let query_id = Self::new_query_id();

        let request_inner = StartResendingEncryptedMessageRequest {
            query_id: query_id.clone(),
            read_cap: read_cap.map(|rc| rc.to_vec()),
            write_cap: write_cap.map(|wc| wc.to_vec()),
            message_box_index: message_box_index.map(|mbi| mbi.to_vec()),
            reply_index,
            envelope_descriptor: envelope_descriptor.to_vec(),
            message_ciphertext: message_ciphertext.to_vec(),
            envelope_hash: envelope_hash.to_vec(),
            no_retry_on_box_id_not_found,
            no_idempotent_box_already_exists,
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("start_resending_encrypted_message".to_string()), request_value);

        // Track in-flight request for replay on reconnect to new daemon instance
        let tracking_key = envelope_hash.to_vec();
        self.in_flight_resends.lock().await.insert(tracking_key.clone(), request.clone());

        // Use direct response routing (like Python's _send_and_wait)
        // This blocks until the daemon sends a reply with matching query_id
        let reply_map = match self.send_and_wait_direct(query_id, request).await {
            Ok(reply) => {
                self.in_flight_resends.lock().await.remove(&tracking_key);
                reply
            }
            Err(e) => {
                self.in_flight_resends.lock().await.remove(&tracking_key);
                return Err(e);
            }
        };

        // Parse the reply
        let reply: StartResendingEncryptedMessageReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        debug!("start_resending_encrypted_message: received reply, error_code={}, plaintext_len={}",
               reply.error_code, reply.plaintext.as_ref().map(|p| p.len()).unwrap_or(0));

        if reply.error_code != 0 {
            return Err(error_code_to_error(reply.error_code));
        }

        Ok(StartResendingResult {
            plaintext: reply.plaintext.unwrap_or_default(),
            courier_identity_hash: reply.courier_identity_hash,
            courier_queue_id: reply.courier_queue_id,
        })
    }

    /// Writes a whole payload, of any size, to a channel using the daemon's
    /// windowed selective-ack (SACK) ARQ. The daemon splits the payload into
    /// as many BACAP boxes as it spans and keeps up to `window` boxes in
    /// flight at once, retransmitting only those whose acknowledgements time
    /// out, so a multi-box payload is no longer serialised one round trip per
    /// box. A `window` of zero asks the daemon to choose a default.
    ///
    /// The daemon does all chunking and encryption; the caller supplies only
    /// the cleartext payload, the write capability, and the start index.
    /// Returns the message box index immediately after the last box written.
    pub async fn write_stream(
        &self,
        write_cap: &[u8],
        start_index: &[u8],
        payload: &[u8],
        window: i64,
    ) -> Result<Vec<u8>, ThinClientError> {
        let query_id = Self::new_query_id();
        let request_inner = WriteStreamRequest {
            query_id: query_id.clone(),
            write_cap: write_cap.to_vec(),
            start_index: start_index.to_vec(),
            payload: payload.to_vec(),
            window,
        };
        let request_value =
            serde_cbor::value::to_value(&request_inner).map_err(|e| ThinClientError::CborError(e))?;
        let mut request = BTreeMap::new();
        request.insert(Value::Text("write_stream".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;
        let reply: WriteStreamReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        debug!("write_stream: received reply, error_code={}, boxes={}", reply.error_code, reply.box_count);
        if reply.error_code != 0 {
            return Err(error_code_to_error(reply.error_code));
        }
        Ok(reply.next_message_box_index.unwrap_or_default())
    }

    /// Reads `box_count` sequential boxes from a channel using the daemon's
    /// windowed selective-ack (SACK) ARQ, the read counterpart of
    /// `write_stream`. The daemon keeps up to `window` boxes in flight,
    /// decrypts each, and reassembles them in order. A `window` of zero asks
    /// the daemon to choose a default. Returns the concatenated payload and the
    /// message box index immediately after the last box read.
    pub async fn read_stream(
        &self,
        read_cap: &[u8],
        start_index: &[u8],
        box_count: u32,
        window: i64,
    ) -> Result<(Vec<u8>, Vec<u8>), ThinClientError> {
        let query_id = Self::new_query_id();
        let request_inner = ReadStreamRequest {
            query_id: query_id.clone(),
            read_cap: read_cap.to_vec(),
            start_index: start_index.to_vec(),
            box_count,
            window,
        };
        let request_value =
            serde_cbor::value::to_value(&request_inner).map_err(|e| ThinClientError::CborError(e))?;
        let mut request = BTreeMap::new();
        request.insert(Value::Text("read_stream".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;
        let reply: ReadStreamReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        debug!("read_stream: received reply, error_code={}, boxes={}", reply.error_code, reply.box_count);
        if reply.error_code != 0 {
            return Err(error_code_to_error(reply.error_code));
        }
        Ok((
            reply.payload.unwrap_or_default(),
            reply.next_message_box_index.unwrap_or_default(),
        ))
    }

    /// Cancels ARQ resending for an encrypted message.
    ///
    /// This method stops the automatic repeat request for a previously started
    /// encrypted message transmission.
    ///
    /// # Arguments
    /// * `envelope_hash` - Hash of the courier envelope to cancel
    ///
    /// # Returns
    /// * `Ok(())` on success
    /// * `Err(ThinClientError)` on failure
    pub async fn cancel_resending_encrypted_message(&self, envelope_hash: &[u8; 32]) -> Result<(), ThinClientError> {
        // Remove from in-flight tracking so it won't be replayed on reconnect
        self.in_flight_resends.lock().await.remove(&envelope_hash.to_vec());

        // If disconnected, just remove from tracking — daemon has no state to cancel
        if !self.is_connected() {
            return Ok(());
        }

        let query_id = Self::new_query_id();

        let request_inner = CancelResendingEncryptedMessageRequest {
            query_id: query_id.clone(),
            envelope_hash: envelope_hash.to_vec(),
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("cancel_resending_encrypted_message".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: CancelResendingEncryptedMessageReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("cancel_resending_encrypted_message failed with error code: {}", reply.error_code)));
        }

        Ok(())
    }

    /// Increments a MessageBoxIndex using the BACAP NextIndex method.
    ///
    /// This method is used when sending multiple messages to different mailboxes using
    /// the same WriteCap or ReadCap. It properly advances the cryptographic state by:
    /// - Incrementing the Idx64 counter
    /// - Deriving new encryption and blinding keys using HKDF
    /// - Updating the HKDF state for the next iteration
    ///
    /// # Arguments
    /// * `message_box_index` - Current message box index to increment
    ///
    /// # Returns
    /// * `Ok(next_message_box_index)` - The incremented message box index
    /// * `Err(ThinClientError)` on failure
    pub async fn next_message_box_index(&self, message_box_index: &[u8]) -> Result<Vec<u8>, ThinClientError> {
        let query_id = Self::new_query_id();

        let request_inner = NextMessageBoxIndexRequest {
            query_id: query_id.clone(),
            message_box_index: message_box_index.to_vec(),
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("next_message_box_index".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: NextMessageBoxIndexReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("next_message_box_index failed with error code: {}", reply.error_code)));
        }

        let next_index = reply.next_message_box_index.ok_or_else(|| ThinClientError::Other("next_message_box_index: next_message_box_index is None".to_string()))?;
        Ok(next_index)
    }

    /// Returns the BACAP `Idx64` counter embedded in a `MessageBoxIndex`.
    ///
    /// Callers that persist `MessageBoxIndex` blobs across sessions can use this
    /// to order or compare two indices — e.g. to detect a duplicate ACK that
    /// would otherwise regress a write-cap's index — without having to peek at
    /// the binary layout themselves. The layout (first 8 bytes little-endian)
    /// is a BACAP implementation detail and must not be relied on outside the
    /// daemon.
    ///
    /// # Arguments
    /// * `message_box_index` - The `MessageBoxIndex` blob whose counter should be returned.
    ///
    /// # Returns
    /// * `Ok(counter)` - The BACAP `Idx64` value.
    /// * `Err(ThinClientError)` on failure.
    pub async fn get_message_box_index_counter(&self, message_box_index: &[u8]) -> Result<u64, ThinClientError> {
        let query_id = Self::new_query_id();

        let request_inner = GetMessageBoxIndexCounterRequest {
            query_id: query_id.clone(),
            message_box_index: message_box_index.to_vec(),
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("get_message_box_index_counter".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: GetMessageBoxIndexCounterReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("get_message_box_index_counter failed with error code: {}", reply.error_code)));
        }

        Ok(reply.counter)
    }

    /// Starts resending a copy command to a courier via ARQ.
    ///
    /// This method instructs a courier to read data from a temporary channel
    /// (identified by the write_cap) and write it to the destination channel.
    /// The command is automatically retransmitted until acknowledged.
    ///
    /// If courier_identity_hash and courier_queue_id are both provided,
    /// the copy command is sent to that specific courier. Otherwise, a
    /// random courier is selected.
    ///
    /// # Arguments
    /// * `write_cap` - Write capability for the temporary channel containing the data
    /// * `courier_identity_hash` - Optional identity hash of a specific courier to use
    /// * `courier_queue_id` - Optional queue ID for the specified courier
    ///
    /// # Returns
    /// * `Ok(())` on success
    /// * `Err(ThinClientError)` on failure
    pub async fn start_resending_copy_command(
        &self,
        write_cap: &[u8],
        courier_identity_hash: Option<&[u8]>,
        courier_queue_id: Option<&[u8]>
    ) -> Result<(), ThinClientError> {
        // Compute write cap hash for in-flight tracking (matches daemon-side hash)
        let tracking_key = Blake2b::<U32>::digest(write_cap).to_vec();

        let query_id = Self::new_query_id();

        let request_inner = StartResendingCopyCommandRequest {
            query_id: query_id.clone(),
            write_cap: write_cap.to_vec(),
            courier_identity_hash: courier_identity_hash.map(|h| h.to_vec()),
            courier_queue_id: courier_queue_id.map(|q| q.to_vec()),
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("start_resending_copy_command".to_string()), request_value);

        // Track in-flight request for replay on reconnect to new daemon instance
        self.in_flight_resends.lock().await.insert(tracking_key.clone(), request.clone());

        let reply_map = match self.send_and_wait_direct(query_id, request).await {
            Ok(reply) => {
                self.in_flight_resends.lock().await.remove(&tracking_key);
                reply
            }
            Err(e) => {
                self.in_flight_resends.lock().await.remove(&tracking_key);
                return Err(e);
            }
        };

        let reply: StartResendingCopyCommandReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if let Some(err) = copy_reply_to_error(&reply) {
            return Err(err);
        }

        Ok(())
    }

    /// Cancels ARQ resending for a copy command.
    ///
    /// This method stops the automatic repeat request (ARQ) for a previously started
    /// copy command.
    ///
    /// # Arguments
    /// * `write_cap_hash` - Hash of the WriteCap used in start_resending_copy_command
    ///
    /// # Returns
    /// * `Ok(())` on success
    /// * `Err(ThinClientError)` on failure
    pub async fn cancel_resending_copy_command(&self, write_cap_hash: &[u8; 32]) -> Result<(), ThinClientError> {
        // Remove from in-flight tracking so it won't be replayed on reconnect
        self.in_flight_resends.lock().await.remove(&write_cap_hash.to_vec());

        // If disconnected, just remove from tracking — daemon has no state to cancel
        if !self.is_connected() {
            return Ok(());
        }

        let query_id = Self::new_query_id();

        let request_inner = CancelResendingCopyCommandRequest {
            query_id: query_id.clone(),
            write_cap_hash: write_cap_hash.to_vec(),
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("cancel_resending_copy_command".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: CancelResendingCopyCommandReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("cancel_resending_copy_command failed with error code: {}", reply.error_code)));
        }

        Ok(())
    }

    /// Packs a payload of arbitrary size (up to 10 MB) into properly sized
    /// `CopyStreamElement` chunks for one destination channel. Each chunk
    /// is a serialised `CopyStreamElement`, ready to be written to a box
    /// via `encrypt_write` followed by `start_resending_encrypted_message`;
    /// the caller marks the boundaries of the stream with the `is_start`
    /// and `is_last` flags.
    ///
    /// This method is stateless: no daemon state is kept between calls,
    /// each invocation runs a fresh encoder and flushes before returning.
    /// The 10 MB cap guards against accidental memory exhaustion.
    ///
    /// Once the chunks have been written to a temporary copy stream, a
    /// copy command (`start_resending_copy_command`) is dispatched to a
    /// courier with the write capability for that temporary stream; the
    /// courier reads the chunks back and writes each envelope to its
    /// destination box.
    ///
    /// Multiple calls can target the same destination stream by passing
    /// `next_dest_index` from the previous result as `dest_start_index`.
    ///
    /// # Arguments
    /// * `payload` - The data to be encoded into courier envelopes
    /// * `dest_write_cap` - Write capability for the destination channel
    /// * `dest_start_index` - Starting index in the destination channel
    /// * `is_start` - Whether this is the first call (sets IsStart on first element)
    /// * `is_last` - Whether this is the last payload in the sequence
    ///
    /// # Returns
    /// * `Ok(CreateEnvelopesResult)` - Contains envelopes and next_dest_index
    /// * `Err(ThinClientError)` on failure
    pub async fn create_courier_envelopes_from_payload(
        &self,
        payload: &[u8],
        dest_write_cap: &[u8],
        dest_start_index: &[u8],
        is_start: bool,
        is_last: bool
    ) -> Result<CreateEnvelopesResult, ThinClientError> {
        let query_id = Self::new_query_id();

        let request_inner = CreateCourierEnvelopesFromPayloadRequest {
            query_id: query_id.clone(),
            payload: payload.to_vec(),
            dest_write_cap: dest_write_cap.to_vec(),
            dest_start_index: dest_start_index.to_vec(),
            is_start,
            is_last,
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("create_courier_envelopes_from_payload".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: CreateCourierEnvelopesFromPayloadReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("create_courier_envelopes_from_payload failed with error code: {}", reply.error_code)));
        }

        Ok(CreateEnvelopesResult {
            envelopes: reply.envelopes.unwrap_or_default().into_iter().map(|b| b.into_vec()).collect(),
            buffer: Vec::new(),
            next_dest_index: reply.next_dest_index,
            next_dest_indices: None,
        })
    }

    /// Packs payloads bound for several destination channels into a single
    /// stream of `CopyStreamElement` chunks. This is more space-efficient
    /// than calling `create_courier_envelopes_from_payload` once per
    /// destination, because the shared encoder runs all envelopes together
    /// rather than padding the final box of each destination independently.
    ///
    /// This method is stateless: the `buffer` argument carries any residual
    /// encoder state across calls in place of daemon-side bookkeeping. Pass
    /// `None` for `buffer` on the first call and the `buffer` returned by
    /// the previous call thereafter; set `is_last` on the final call so the
    /// encoder flushes its tail.
    ///
    /// # Arguments
    /// * `destinations` - List of (payload, write_cap, start_index) tuples
    /// * `is_start` - Whether this is the first call in the sequence
    /// * `is_last` - Whether this is the last set of payloads in the sequence
    /// * `buffer` - Residual encoder buffer from a previous call, or None
    ///
    /// # Returns
    /// * `Ok(CreateEnvelopesResult)` - Contains envelopes and buffer for next call
    /// * `Err(ThinClientError)` on failure
    pub async fn create_courier_envelopes_from_multi_payload(
        &self,
        destinations: Vec<(&[u8], &[u8], &[u8])>,
        is_start: bool,
        is_last: bool,
        buffer: Option<Vec<u8>>,
    ) -> Result<CreateEnvelopesResult, ThinClientError> {
        let query_id = Self::new_query_id();

        let destinations_inner: Vec<EnvelopeDestination> = destinations
            .into_iter()
            .map(|(payload, write_cap, start_index)| EnvelopeDestination {
                payload: payload.to_vec(),
                write_cap: write_cap.to_vec(),
                start_index: start_index.to_vec(),
            })
            .collect();

        let request_inner = CreateCourierEnvelopesFromPayloadsRequest {
            query_id: query_id.clone(),
            destinations: destinations_inner,
            is_start,
            is_last,
            buffer,
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("create_courier_envelopes_from_multi_payload".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: CreateCourierEnvelopesFromPayloadsReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("create_courier_envelopes_from_multi_payload failed with error code: {}", reply.error_code)));
        }

        Ok(CreateEnvelopesResult {
            envelopes: reply.envelopes.unwrap_or_default().into_iter().map(|b| b.into_vec()).collect(),
            buffer: reply.buffer.unwrap_or_default(),
            next_dest_index: None,
            next_dest_indices: reply.next_dest_indices.map(|v| v.into_iter().map(|b| b.into_vec()).collect()),
        })
    }

    /// Packs tombstones for a consecutive range of destination boxes into
    /// `CopyStreamElement` chunks. The chunks are written to a temporary
    /// copy stream and then dispatched as a copy command; the courier
    /// applies all the tombstones atomically, which is the natural way to
    /// retire a range of boxes as part of the same copy transaction that
    /// writes their successors.
    ///
    /// This method is stateless: the `buffer` argument carries any residual
    /// encoder state across calls in place of daemon-side bookkeeping.
    /// Pass `None` for `buffer` on the first call and the `buffer`
    /// returned by the previous call thereafter; set `is_last` on the
    /// final call so the encoder flushes its tail.
    ///
    /// # Arguments
    /// * `dest_write_cap` - Write capability for the destination channel
    /// * `dest_start_index` - Starting index in the destination channel
    /// * `max_count` - Number of tombstones to create
    /// * `is_start` - Whether this is the first call in the sequence
    /// * `is_last` - Whether this is the last call in the sequence
    /// * `buffer` - Residual encoder buffer from a previous call, or None
    ///
    /// # Returns
    /// * `Ok(CreateEnvelopesResult)` - Contains envelopes, buffer, and next_dest_index
    /// * `Err(ThinClientError)` on failure
    pub async fn create_courier_envelopes_from_tombstone_range(
        &self,
        dest_write_cap: &[u8],
        dest_start_index: &[u8],
        max_count: u32,
        is_start: bool,
        is_last: bool,
        buffer: Option<Vec<u8>>,
    ) -> Result<CreateEnvelopesResult, ThinClientError> {
        let query_id = Self::new_query_id();

        let request_inner = CreateCourierEnvelopesFromTombstoneRangeRequest {
            query_id: query_id.clone(),
            dest_write_cap: dest_write_cap.to_vec(),
            dest_start_index: dest_start_index.to_vec(),
            max_count,
            is_start,
            is_last,
            buffer,
        };

        let request_value = serde_cbor::value::to_value(&request_inner)
            .map_err(|e| ThinClientError::CborError(e))?;

        let mut request = BTreeMap::new();
        request.insert(Value::Text("create_courier_envelopes_from_tombstone_range".to_string()), request_value);

        let reply_map = self.send_and_wait_direct(query_id, request).await?;

        let reply: CreateCourierEnvelopesFromTombstoneRangeReply = serde_cbor::value::from_value(Value::Map(reply_map))
            .map_err(|e| ThinClientError::CborError(e))?;

        if reply.error_code != 0 {
            return Err(ThinClientError::Other(format!("create_courier_envelopes_from_tombstone_range failed with error code: {}", reply.error_code)));
        }

        Ok(CreateEnvelopesResult {
            envelopes: reply.envelopes.unwrap_or_default().into_iter().map(|b| b.into_vec()).collect(),
            buffer: reply.buffer.unwrap_or_default(),
            next_dest_index: reply.next_dest_index,
            next_dest_indices: None,
        })
    }

}

/// A single tombstone envelope ready to be sent.
#[derive(Debug, Clone)]
pub struct TombstoneEnvelope {
    /// The encrypted tombstone payload.
    pub message_ciphertext: Vec<u8>,
    /// The envelope descriptor.
    pub envelope_descriptor: Vec<u8>,
    /// The envelope hash for cancellation.
    pub envelope_hash: Vec<u8>,
    /// The box index this envelope is for.
    pub box_index: Vec<u8>,
}

/// Result of a tombstone_range operation.
#[derive(Debug)]
pub struct TombstoneRangeResult {
    /// List of tombstone envelopes ready to be sent.
    pub envelopes: Vec<TombstoneEnvelope>,
    /// The next MessageBoxIndex after the last processed.
    pub next: Vec<u8>,
    /// Error message if the operation failed partway through.
    pub error: Option<String>,
}

impl ThinClient {
    /// Prepares the encrypted envelopes needed to tombstone a consecutive
    /// range of pigeonhole boxes beginning at the supplied
    /// `MessageBoxIndex`. A tombstone is a signed empty payload that the
    /// replica recognises as a deletion marker; the daemon constructs one
    /// by signing rather than encrypting whenever `encrypt_write` is
    /// invoked with an empty plaintext.
    ///
    /// This method does not itself touch the network: it returns the
    /// envelopes for the caller to dispatch one by one, typically via
    /// `start_resending_encrypted_message`. To tombstone a single box,
    /// pass `max_count = 1`.
    ///
    /// If an error occurs while creating envelopes, a partial result is
    /// returned containing the envelopes created so far together with the
    /// next index and an error message, so the caller may resume or
    /// surface the failure as appropriate.
    ///
    /// # Arguments
    /// * `write_cap` - Write capability for the boxes
    /// * `start` - Starting MessageBoxIndex
    /// * `max_count` - Maximum number of boxes to tombstone
    ///
    /// # Returns
    /// * `TombstoneRangeResult` containing the envelopes and next index
    pub async fn tombstone_range(
        &self,
        write_cap: &[u8],
        start: &[u8],
        max_count: u32
    ) -> TombstoneRangeResult {
        if max_count == 0 {
            return TombstoneRangeResult {
                envelopes: Vec::new(),
                next: start.to_vec(),
                error: None,
            };
        }

        let mut cur = start.to_vec();
        let mut envelopes: Vec<TombstoneEnvelope> = Vec::with_capacity(max_count as usize);

        while (envelopes.len() as u32) < max_count {
            match self.encrypt_write(&[], write_cap, &cur).await {
                Ok(result) => {
                    envelopes.push(TombstoneEnvelope {
                        message_ciphertext: result.message_ciphertext,
                        envelope_descriptor: result.envelope_descriptor,
                        envelope_hash: result.envelope_hash.to_vec(),
                        box_index: cur.clone(),
                    });
                    cur = result.next_message_box_index;
                }
                Err(e) => {
                    let count = envelopes.len();
                    return TombstoneRangeResult {
                        envelopes,
                        next: cur,
                        error: Some(format!("Error creating tombstone at index {}: {:?}", count, e)),
                    };
                }
            }
        }

        TombstoneRangeResult {
            envelopes,
            next: cur,
            error: None,
        }
    }
}

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

    fn map_keys(value: Value) -> Vec<String> {
        match value {
            Value::Map(m) => m
                .into_keys()
                .filter_map(|k| match k {
                    Value::Text(s) => Some(s),
                    _ => None,
                })
                .collect(),
            _ => panic!("expected a CBOR map"),
        }
    }

    // The daemon decodes these requests by CBOR field name; a rename here that
    // drifts from the Go `cbor:"..."` tags would silently break interop, so we
    // pin the wire field names.
    #[test]
    fn write_stream_request_field_names() {
        let req = WriteStreamRequest {
            query_id: vec![1, 2, 3],
            write_cap: vec![4, 5],
            start_index: vec![6, 7],
            payload: vec![8, 9],
            window: 16,
        };
        let keys = map_keys(serde_cbor::value::to_value(&req).unwrap());
        for expected in ["query_id", "write_cap", "start_index", "payload", "window"] {
            assert!(keys.iter().any(|k| k == expected), "missing field {expected}, got {keys:?}");
        }
    }

    #[test]
    fn read_stream_request_field_names() {
        let req = ReadStreamRequest {
            query_id: vec![1],
            read_cap: vec![2],
            start_index: vec![3],
            box_count: 4,
            window: 0,
        };
        let keys = map_keys(serde_cbor::value::to_value(&req).unwrap());
        for expected in ["query_id", "read_cap", "start_index", "box_count", "window"] {
            assert!(keys.iter().any(|k| k == expected), "missing field {expected}, got {keys:?}");
        }
    }
}