polyc-state-connect 2026.8.3

//! Capability-specific agent-task client.
//!
//! Async end to end. The generated client is async, every method here awaits
//! it, and nothing in this family reaches for a blocking bridge or a blocking
//! pool — a typed State family has no reason to, and #2140 is what happens
//! when one does.

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    error::StateError,
    id::{CommandId, NamespaceId},
    receipt::Receipt,
    revision::Revision,
    tasks::{ContextId, ContextIndex, EdgeId, TaskCommand, TaskFact, TaskId, TaskPage, TaskRecord},
};

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

use super::wire::{index_from_wire, metadata_to_wire, operation_to_wire, record_from_wire};

/// Client bound to one task namespace and no other State capability.
pub struct TaskClient<T> {
    inner: pb::StateTaskServiceClient<T>,
    namespace: NamespaceId,
}

impl<T> TaskClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a namespace-bound client.
    #[must_use]
    pub fn new(transport: T, config: ClientConfig, namespace: NamespaceId) -> Self {
        Self {
            inner: pb::StateTaskServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_WIRE_MESSAGE_BYTES),
            ),
            namespace,
        }
    }

    fn fallback(attempted: usize) -> TransportFallback {
        TransportFallback::new(
            polyc_state::tasks::family(),
            MAX_WIRE_MESSAGE_BYTES as u64,
            attempted as u64,
        )
    }

    /// Commits one typed task mutation.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure. Cancelling a task that
    /// already finished comes back as a refusal, never as a quiet success.
    pub async fn transact(
        &self,
        declared: &DeclaredCall,
        command: &TaskCommand,
    ) -> Result<Receipt, StateError> {
        if command.metadata().scope().namespace() != &self.namespace {
            return Err(StateError::Denied {
                family: polyc_state::tasks::family(),
            });
        }
        let request = pb::TransactStateTaskRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            metadata: buffa::MessageField::some(metadata_to_wire(command)),
            operation: buffa::MessageField::some(operation_to_wire(command.operation())),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .transact_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Kernel::<Receipt>::try_from(reply.receipt.into_option().ok_or_else(|| {
            StateError::Malformed {
                field: "receipt".into(),
                reason: "a successful task mutation returns a receipt".into(),
            }
        })?)
        .map(Kernel::into_inner)
    }

    /// Reads one task by identity.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn task(
        &self,
        declared: &DeclaredCall,
        owner: &EdgeId,
        id: &TaskId,
    ) -> Result<TaskFact<Option<TaskRecord>>, StateError> {
        let request = pb::GetStateTaskRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            task_id: id.as_str().to_owned(),
            owner_edge_id: owner.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_task_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(TaskFact::observed(
            reply.task.into_option().map(record_from_wire).transpose()?,
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one owner's bounded index for one context.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn context_index(
        &self,
        declared: &DeclaredCall,
        owner: &EdgeId,
        context_id: &ContextId,
    ) -> Result<TaskFact<ContextIndex>, StateError> {
        let request = pb::GetStateTaskContextIndexRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            context_id: context_id.as_str().to_owned(),
            owner_edge_id: owner.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_context_index_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        Ok(TaskFact::observed(
            index_from_wire(
                reply
                    .index
                    .into_option()
                    .ok_or_else(|| StateError::Malformed {
                        field: "index".into(),
                        reason: "a task index reply carries its index".into(),
                    })?,
            ),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Reads one bounded page of one owner's tasks in one context.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure, including a refusal for a
    /// zero or oversized page.
    pub async fn list_by_context(
        &self,
        declared: &DeclaredCall,
        owner: &EdgeId,
        context_id: &ContextId,
        after: Option<&TaskId>,
        page_size: usize,
    ) -> Result<TaskFact<TaskPage>, StateError> {
        let request = pb::ListStateTasksRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            context_id: context_id.as_str().to_owned(),
            after: after.map(|id| id.as_str().to_owned()),
            page_size: u32::try_from(page_size).unwrap_or(u32::MAX),
            owner_edge_id: owner.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .list_tasks_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let tasks = reply
            .tasks
            .into_iter()
            .map(record_from_wire)
            .collect::<Result<Vec<_>, _>>()?;
        Ok(TaskFact::observed(
            TaskPage::new(tasks, reply.next_after.map(TaskId::new)),
            Revision::new(reply.snapshot_revision),
            reply.entry_revision.map(Revision::new),
        ))
    }

    /// Settles an ambiguous command from its durable receipt.
    ///
    /// # Errors
    ///
    /// Returns a typed State or transport failure.
    pub async fn committed_receipt(
        &self,
        declared: &DeclaredCall,
        command_id: &CommandId,
    ) -> Result<Option<Receipt>, StateError> {
        let request = pb::GetStateTaskReceiptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: self.namespace.as_str().to_owned(),
            command_id: command_id.as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .get_receipt_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        reply
            .receipt
            .into_option()
            .map(|value| Kernel::<Receipt>::try_from(value).map(Kernel::into_inner))
            .transpose()
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use std::{
        pin::Pin,
        task::{Context, Poll},
        time::Duration,
    };

    use bytes::Bytes;
    use connectrpc::{
        client::{BoxFuture, ClientBody},
        http_body::{Body, Frame},
    };

    use super::*;

    struct NeverBody;

    impl Body for NeverBody {
        type Data = Bytes;
        type Error = std::io::Error;

        fn poll_frame(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
        ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
            Poll::Pending
        }
    }

    #[derive(Clone, Copy)]
    struct StalledTransport;

    impl ClientTransport for StalledTransport {
        type ResponseBody = NeverBody;
        type Error = std::io::Error;

        fn send(
            &self,
            _request: http::Request<ClientBody>,
        ) -> BoxFuture<'static, Result<http::Response<Self::ResponseBody>, Self::Error>> {
            Box::pin(std::future::pending())
        }
    }

    /// Drives a real generated unary call over a transport that never returns.
    /// The outer second is only the test watchdog; the State call must refuse
    /// from its own 20 ms declaration.
    #[tokio::test]
    async fn a_stalled_transport_refuses_within_the_declared_budget() {
        let client = TaskClient::new(
            StalledTransport,
            ClientConfig::new("http://state.invalid".parse().unwrap()),
            NamespaceId::new("deadline-test"),
        );
        let declared = DeclaredCall::bounded(crate::state_audience(), Duration::from_millis(20));
        let result = tokio::time::timeout(
            Duration::from_secs(1),
            client.task(&declared, &EdgeId::new("edge-1"), &TaskId::new("stalled")),
        )
        .await
        .expect("the unary client's own deadline must fire");

        assert!(
            matches!(result, Err(StateError::DeadlineExpired { .. })),
            "a stalled authority fails closed as a deadline, got {result:?}"
        );
    }
}