polyc-state-connect 2026.8.3

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    error::StateError,
    model_attempt::{
        DispatchDisposition, ModelAttemptLifecycle, ModelAttemptReceipt, ModelAttemptRequest,
    },
};

use crate::{
    MAX_WIRE_MESSAGE_BYTES,
    error::{TransportFallback, from_connect_error},
    trace::bounded_traced_options,
    wire::{DeclaredCall, Kernel},
};

use super::wire::{lifecycle_to_wire, receipt_from_wire, request_to_wire};

/// Capability-specific State client for durable model attempts.
pub struct ModelAttemptClient<T> {
    inner: pb::StateModelAttemptServiceClient<T>,
}

impl<T> ModelAttemptClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a client for one State listener.
    #[must_use]
    pub fn new(transport: T, config: ClientConfig) -> Self {
        Self {
            inner: pb::StateModelAttemptServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_WIRE_MESSAGE_BYTES),
            ),
        }
    }
    fn fallback(attempted: usize) -> TransportFallback {
        TransportFallback::new(
            polyc_state::model_attempt::family(),
            MAX_WIRE_MESSAGE_BYTES as u64,
            attempted as u64,
        )
    }
    /// Reserves one durable model attempt.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport refusal.
    pub async fn reserve(
        &self,
        declared: &DeclaredCall,
        request: &ModelAttemptRequest,
    ) -> Result<ModelAttemptReceipt, StateError> {
        let message = pb::ReserveModelAttemptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            request: buffa::MessageField::some(request_to_wire(request)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&message) as usize;
        let reply = self
            .inner
            .reserve_with_options(message, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        receipt_from_wire(
            reply
                .receipt
                .into_option()
                .ok_or_else(|| StateError::Malformed {
                    field: "receipt".to_owned(),
                    reason: "reserve returns its receipt".to_owned(),
                })?,
        )
    }
    /// Records whether this request may begin provider I/O.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport refusal. A reply this caller cannot
    /// read reports an unknown outcome, never a refusal: the call returned, so
    /// the dispatch is already recorded.
    pub async fn dispatch(
        &self,
        declared: &DeclaredCall,
        request: &ModelAttemptRequest,
    ) -> Result<DispatchDisposition, StateError> {
        let message = pb::DispatchModelAttemptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            request: buffa::MessageField::some(request_to_wire(request)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&message) as usize;
        let reply = self
            .inner
            .dispatch_with_options(message, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        // The call returned, so the authority already recorded the dispatch.
        // A reply this caller cannot read does not undo that. Reporting it as
        // malformed would make it terminal, and a terminal dispatch failure
        // becomes a durable turn failure beside a marker nobody settles
        // (`#2638`). The outcome is unknown instead, which is what it is.
        //
        // A skewed record version reaches this same path, so a rolling upgrade
        // yields unknown outcomes rather than false definite ones (`#2639`).
        let unreadable_reply = || StateError::Unavailable {
            family: polyc_state::model_attempt::family(),
            reach: polyc_state::error::OutageReach::PossiblyApplied,
        };
        let receipt = receipt_from_wire(reply.receipt.into_option().ok_or_else(unreadable_reply)?)
            .map_err(|_| unreadable_reply())?;
        Ok(if reply.begin {
            DispatchDisposition::Begin
        } else {
            DispatchDisposition::Existing(receipt)
        })
    }
    /// Advances one durable attempt lifecycle.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport refusal.
    pub async fn transition(
        &self,
        declared: &DeclaredCall,
        request: &ModelAttemptRequest,
        next: ModelAttemptLifecycle,
    ) -> Result<ModelAttemptReceipt, StateError> {
        let message = pb::TransitionModelAttemptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            request: buffa::MessageField::some(request_to_wire(request)),
            next: buffa::EnumValue::Known(lifecycle_to_wire(next)),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&message) as usize;
        let reply = self
            .inner
            .transition_with_options(message, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        receipt_from_wire(
            reply
                .receipt
                .into_option()
                .ok_or_else(|| StateError::Malformed {
                    field: "receipt".to_owned(),
                    reason: "transition returns its receipt".to_owned(),
                })?,
        )
    }
}