polyc-state-connect 2026.9.0

//! Capability-specific client for Query's audit trail.

use std::collections::BTreeSet;

use connectrpc::client::{ClientConfig, ClientTransport};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    command::{CommandEnvelope, CommandMetadata},
    digest::ContentDigest,
    error::StateError,
    id::NamespaceId,
    page::{Page, PageCompleteness, Positioned, ReadStart},
    query_audit::{
        AuditPhase, BeginOutcome, BeginQueryAudit, ListUnmatchedIntents, MAX_AUDIT_PAGE_BYTES,
        MAX_UNMATCHED_INTENTS_PER_PAGE, QueryAudit, QueryAuditError, QueryCompletion, QueryId,
        ReadQueryAudit, RecordedCompletionInput, SourceSnapshot,
    },
    receipt::Receipt,
};

/// Opaque proof that the trusted State listener durably recorded one intent.
///
/// The value is deliberately non-cloneable and has no public constructor. It
/// proves ordering under the authenticated, crash-fault State model. Receipts
/// are not signatures. This is not Byzantine or cryptographic proof of remote
/// durability.
///
/// Safe downstream code cannot mint it:
///
/// ```compile_fail
/// use polyc_state_connect::query_audit::RemoteExecutionPermit;
/// let _ = RemoteExecutionPermit {
///     query: todo!(),
///     namespace: todo!(),
///     source: todo!(),
///     receipt: todo!(),
/// };
/// ```
///
/// It is also one-shot rather than cloneable:
///
/// ```compile_fail
/// use polyc_state_connect::query_audit::RemoteExecutionPermit;
/// let permit: RemoteExecutionPermit = todo!();
/// let _second = permit.clone();
/// ```
#[derive(Debug, PartialEq, Eq)]
pub struct RemoteExecutionPermit {
    query: QueryId,
    namespace: NamespaceId,
    source: SourceSnapshot,
    receipt: Receipt,
}

impl RemoteExecutionPermit {
    /// Returns the query whose durable intent was acknowledged.
    #[must_use]
    pub const fn query(&self) -> &QueryId {
        &self.query
    }

    /// Returns the tenant namespace whose durable intent was acknowledged.
    #[must_use]
    pub const fn namespace(&self) -> &NamespaceId {
        &self.namespace
    }

    /// Returns the source premises carried by the acknowledged intent.
    #[must_use]
    pub const fn source(&self) -> &SourceSnapshot {
        &self.source
    }

    /// Returns the exact receipt validated by the client.
    #[must_use]
    pub const fn receipt(&self) -> &Receipt {
        &self.receipt
    }

    /// Returns the canonical completion payload without consuming this permit.
    #[must_use]
    pub fn completion_canonical_bytes(&self, completion: &QueryCompletion) -> Vec<u8> {
        RecordedCompletionInput::canonical_bytes_for(
            &self.query,
            &self.namespace,
            &self.source,
            completion,
        )
    }

    /// Consumes this one-shot permit into a retryable completion command.
    ///
    /// # Errors
    ///
    /// Returns a malformed or bounds refusal when the completion drifts from
    /// the intent source premises or exceeds the query-audit payload bound.
    pub fn into_completion(
        self,
        completion: QueryCompletion,
        digest: ContentDigest,
        envelope: CommandEnvelope,
    ) -> Result<RemoteCompleteQueryAudit, QueryAuditError> {
        let input = RecordedCompletionInput::new(
            self.query,
            self.namespace,
            self.source,
            completion,
            digest,
            envelope,
        );
        input.validate_syntax()?;
        Ok(RemoteCompleteQueryAudit { input })
    }
}

/// Retryable remote completion built by consuming one remote permit.
///
/// Its fields are private. A caller may clone the finished command to settle an
/// ambiguous response. It cannot address a second completion from that permit.
///
/// ```compile_fail
/// use polyc_state_connect::query_audit::RemoteCompleteQueryAudit;
/// let _ = RemoteCompleteQueryAudit { input: todo!() };
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RemoteCompleteQueryAudit {
    input: RecordedCompletionInput,
}

impl RemoteCompleteQueryAudit {
    fn validate_syntax(&self) -> Result<(), QueryAuditError> {
        self.input.validate_syntax().map_err(QueryAuditError::from)
    }

    /// Returns the exact completion metadata used for receipt settlement.
    #[must_use]
    pub const fn metadata(&self) -> &CommandMetadata {
        self.input.metadata()
    }

    /// Returns the addressed query.
    #[must_use]
    pub const fn query(&self) -> &QueryId {
        self.input.query()
    }

    /// Returns the addressed tenant namespace.
    #[must_use]
    pub const fn namespace(&self) -> &NamespaceId {
        self.input.namespace()
    }

    /// Returns the source premises recorded by the intent.
    #[must_use]
    pub const fn intent_source(&self) -> &SourceSnapshot {
        self.input.intent_source()
    }

    /// Returns the recorded outcome.
    #[must_use]
    pub const fn completion(&self) -> &QueryCompletion {
        self.input.completion()
    }
}

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

/// A client that can write and inspect query audit records, and nothing else.
pub struct QueryAuditClient<T> {
    inner: pb::StateQueryAuditServiceClient<T>,
}

impl<T> QueryAuditClient<T>
where
    T: ClientTransport,
    <T::ResponseBody as connectrpc::http_body::Body>::Error: std::fmt::Display,
{
    /// Builds a client under the query-audit wire bound.
    pub fn new(transport: T, config: ClientConfig) -> Self {
        Self {
            inner: pb::StateQueryAuditServiceClient::new(
                transport,
                config.with_default_max_message_size(MAX_QUERY_AUDIT_WIRE_MESSAGE_BYTES),
            ),
        }
    }

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

    /// Records or settles one durable intent. A new intent returns its sole
    /// permit. An exact retry returns recorded evidence without a permit.
    ///
    /// # Errors
    ///
    /// Returns the exact query-audit outcome carried by State. Any error means
    /// no permit and therefore no authorized execution.
    pub async fn begin(
        &self,
        declared: &DeclaredCall,
        command: &BeginQueryAudit,
    ) -> Result<BeginOutcome<RemoteExecutionPermit>, QueryAuditError> {
        command.validate_syntax()?;
        let metadata = command.metadata();
        let request = pb::BeginQueryAuditRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            query: command.query().as_str().to_owned(),
            namespace: command.namespace().as_str().to_owned(),
            requester: command.requester().as_str().to_owned(),
            shape: command.shape().as_bytes().to_vec(),
            source: buffa::MessageField::some(pb::QuerySourceSnapshot::from(Kernel(
                command.source(),
            ))),
            digest: metadata.digest().as_bytes().to_vec(),
            purpose: metadata.envelope().purpose().as_str().to_owned(),
            command_audience: metadata.envelope().audience().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .begin_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let receipt = receipt(reply.receipt)?;
        validate_receipt(&receipt, command.metadata())?;
        if receipt.is_new() {
            Ok(BeginOutcome::Granted(RemoteExecutionPermit {
                query: command.query().clone(),
                namespace: command.namespace().clone(),
                source: command.source().clone(),
                receipt,
            }))
        } else {
            Ok(BeginOutcome::AlreadyRecorded(Box::new(receipt)))
        }
    }

    /// Durably records the outcome of an already-permitted query.
    ///
    /// # Errors
    ///
    /// Returns the exact family outcome. A dropped or ambiguous call is
    /// settled by retrying this same command or calling
    /// [`Self::completion_receipt`].
    pub async fn complete(
        &self,
        declared: &DeclaredCall,
        command: &RemoteCompleteQueryAudit,
    ) -> Result<Receipt, QueryAuditError> {
        command.validate_syntax()?;
        let request = pb::CompleteQueryAuditRequest::from(Kernel((declared, &command.input)));
        let attempted = buffa::Message::encoded_len(&request) as usize;
        let reply = self
            .inner
            .complete_with_options(request, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let receipt = receipt(reply.receipt)?;
        validate_receipt(&receipt, command.metadata())?;
        Ok(receipt)
    }

    /// Settles an ambiguous completion against the exact retryable command.
    ///
    /// # Errors
    ///
    /// Returns a malformed-response refusal for fresh lookup evidence, crossed
    /// metadata, or durable completion semantics that differ from the attempt.
    pub async fn completion_receipt(
        &self,
        declared: &DeclaredCall,
        command: &RemoteCompleteQueryAudit,
    ) -> Result<Option<Receipt>, QueryAuditError> {
        let receipt = self
            .receipt(
                declared,
                command.query(),
                command.namespace(),
                AuditPhase::Completion,
            )
            .await?;
        let Some(receipt) = receipt else {
            return Ok(None);
        };
        validate_receipt(&receipt, command.metadata())?;
        if !receipt.is_deduplicated() {
            return Err(malformed_response(
                "receipt.disposition",
                "a receipt lookup returns recorded evidence",
            ));
        }
        let audit = self
            .audit(
                declared,
                &ReadQueryAudit::new(command.query().clone(), command.namespace().clone()),
            )
            .await?
            .ok_or_else(|| {
                malformed_response(
                    "audit",
                    "a completion receipt belongs to an existing audit trail",
                )
            })?;
        if audit.intent().source() != command.intent_source()
            || audit.completion() != Some(command.completion())
        {
            return Err(malformed_response(
                "audit.completion",
                "the durable completion equals the exact attempted completion",
            ));
        }
        Ok(Some(receipt))
    }

    /// Reads one query's whole audit trail.
    ///
    /// # Errors
    ///
    /// Returns the exact family outcome or a malformed-response outcome if a
    /// successful listener sent an invalid record.
    pub async fn audit(
        &self,
        declared: &DeclaredCall,
        request: &ReadQueryAudit,
    ) -> Result<Option<QueryAudit>, QueryAuditError> {
        validate_read_scope(request.query(), request.namespace())?;
        let wire = pb::GetQueryAuditRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            query: request.query().as_str().to_owned(),
            namespace: request.namespace().as_str().to_owned(),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .get_audit_with_options(wire, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let audit = reply
            .audit
            .into_option()
            .map(|audit| {
                Kernel::<QueryAudit>::try_from(audit)
                    .map(Kernel::into_inner)
                    .map_err(QueryAuditError::from)
            })
            .transpose()?;
        if audit.as_ref().is_some_and(|audit| {
            audit.query() != request.query() || audit.namespace() != request.namespace()
        }) {
            return Err(malformed_response(
                "audit.intent.scope",
                "must equal the query and namespace requested",
            ));
        }
        Ok(audit)
    }

    /// Lists one bounded page of unmatched intents.
    ///
    /// # Errors
    ///
    /// Returns the exact family outcome, including the page bound.
    pub async fn unmatched(
        &self,
        declared: &DeclaredCall,
        request: &ListUnmatchedIntents,
    ) -> Result<Page<QueryAudit>, QueryAuditError> {
        validate_unmatched_request(request)?;
        let wire = pb::ListUnmatchedQueryAuditsRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            namespace: request.namespace().as_str().to_owned(),
            page: buffa::MessageField::some(Kernel(request.page()).into()),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .list_unmatched_with_options(wire, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        let page = reply.page.into_option().ok_or_else(|| {
            QueryAuditError::from(StateError::Malformed {
                field: "page".to_owned(),
                reason: "a successful listing carries its page".to_owned(),
            })
        })?;
        let page = Kernel::<Page<QueryAudit>>::try_from(page)
            .map(Kernel::into_inner)
            .map_err(QueryAuditError::from)?;
        validate_unmatched_page(request, &page)?;
        Ok(page)
    }

    /// Retrieves the durable receipt for one phase for diagnostics.
    ///
    /// This phase lookup cannot settle an ambiguous content write because the
    /// lookup carries no attempted digest. Use [`Self::completion_receipt`] for
    /// completion settlement.
    ///
    /// # Errors
    ///
    /// Returns the exact family outcome. Absence means that phase never
    /// committed, not that State could not determine its outcome.
    pub async fn receipt(
        &self,
        declared: &DeclaredCall,
        query: &QueryId,
        namespace: &polyc_state::id::NamespaceId,
        phase: AuditPhase,
    ) -> Result<Option<Receipt>, QueryAuditError> {
        validate_read_scope(query, namespace)?;
        let wire = pb::GetQueryAuditReceiptRequest {
            context: buffa::MessageField::some(Kernel(declared).into()),
            query: query.as_str().to_owned(),
            namespace: namespace.as_str().to_owned(),
            completion: matches!(phase, AuditPhase::Completion),
            __buffa_unknown_fields: buffa::UnknownFields::default(),
        };
        let attempted = buffa::Message::encoded_len(&wire) as usize;
        let reply = self
            .inner
            .get_receipt_with_options(wire, bounded_traced_options(declared))
            .await
            .map_err(|error| from_connect_error(&error, &Self::fallback(attempted)))?
            .into_owned();
        reply
            .receipt
            .into_option()
            .map(|receipt| {
                let receipt = Kernel::<Receipt>::try_from(receipt)
                    .map(Kernel::into_inner)
                    .map_err(QueryAuditError::from)?;
                validate_retrieved_receipt(&receipt, query, namespace, phase)?;
                Ok(receipt)
            })
            .transpose()
    }
}

fn malformed_response(field: &str, reason: &str) -> QueryAuditError {
    StateError::Malformed {
        field: field.to_owned(),
        reason: reason.to_owned(),
    }
    .into()
}

fn validate_read_scope(
    query: &QueryId,
    namespace: &polyc_state::id::NamespaceId,
) -> Result<(), QueryAuditError> {
    if query.is_empty() {
        return Err(malformed_response(
            "query",
            "a query identity may not be empty",
        ));
    }
    if namespace.as_str().is_empty() {
        return Err(malformed_response(
            "namespace",
            "a tenant namespace may not be empty",
        ));
    }
    Ok(())
}

fn validate_unmatched_request(request: &ListUnmatchedIntents) -> Result<(), QueryAuditError> {
    if request.namespace().as_str().is_empty() {
        return Err(malformed_response(
            "namespace",
            "a tenant namespace may not be empty",
        ));
    }
    if request.page().limit() == 0 {
        return Err(malformed_response("page.limit", "must make progress"));
    }
    if request.page().limit() > MAX_UNMATCHED_INTENTS_PER_PAGE {
        return Err(StateError::BoundsExceeded {
            bound: polyc_state::error::BoundKind::PageRecords,
            limit: u64::from(MAX_UNMATCHED_INTENTS_PER_PAGE),
            requested: u64::from(request.page().limit()),
        }
        .into());
    }
    let ReadStart::Resume(cursor) = request.page().start() else {
        return Err(malformed_response(
            "page.start",
            "query-audit listings resume from an unbound cursor",
        ));
    };
    if cursor.snapshot().is_some() {
        return Err(malformed_response(
            "start.snapshot",
            "query-audit listings have no snapshot binding",
        ));
    }
    Ok(())
}

fn validate_receipt(receipt: &Receipt, command: &CommandMetadata) -> Result<(), QueryAuditError> {
    if !receipt.answers(command) {
        return Err(malformed_response(
            "receipt.command",
            "must answer the exact command identity and digest",
        ));
    }
    validate_receipt_contract(receipt)
}

fn validate_retrieved_receipt(
    receipt: &Receipt,
    query: &QueryId,
    namespace: &polyc_state::id::NamespaceId,
    phase: AuditPhase,
) -> Result<(), QueryAuditError> {
    if receipt.command_id() != &polyc_state::query_audit::command_id(query, namespace, phase) {
        return Err(malformed_response(
            "receipt.command_id",
            "must name the requested namespace, query, and phase",
        ));
    }
    validate_receipt_contract(receipt)
}

fn validate_receipt_contract(receipt: &Receipt) -> Result<(), QueryAuditError> {
    if receipt.family() != &polyc_state::query_audit::family() {
        return Err(malformed_response(
            "receipt.family",
            "must name the query-audit operation family",
        ));
    }
    if receipt.consistency() != polyc_state::query_audit::AUDIT_CONSISTENCY {
        return Err(malformed_response(
            "receipt.consistency",
            "must name the query-audit consistency contract",
        ));
    }
    if receipt.fence().is_some() {
        return Err(malformed_response(
            "receipt.fence",
            "query-audit commands never carry a fencing token",
        ));
    }
    Ok(())
}

fn validate_unmatched_page(
    request: &ListUnmatchedIntents,
    page: &Page<QueryAudit>,
) -> Result<(), QueryAuditError> {
    validate_unmatched_request(request)?;
    let ReadStart::Resume(requested) = request.page().start() else {
        unreachable!("validate_unmatched_request admits only resume cursors")
    };
    if page.consistency() != polyc_state::query_audit::AUDIT_CONSISTENCY {
        return Err(malformed_response(
            "page.consistency",
            "must name the query-audit consistency contract",
        ));
    }
    let requested_limit = usize::try_from(request.page().limit()).unwrap_or(usize::MAX);
    let family_limit = usize::try_from(MAX_UNMATCHED_INTENTS_PER_PAGE).unwrap_or(usize::MAX);
    if page.len() > requested_limit || page.len() > family_limit {
        return Err(malformed_response(
            "page.records",
            "must not exceed the requested or family record bound",
        ));
    }
    if page.records().iter().any(|audit| !audit.is_unmatched()) {
        return Err(malformed_response(
            "page.records.completion",
            "an unmatched listing cannot carry a completed audit",
        ));
    }
    if page
        .records()
        .iter()
        .any(|audit| audit.namespace() != request.namespace())
    {
        return Err(malformed_response(
            "page.records.namespace",
            "every unmatched intent must belong to the requested namespace",
        ));
    }

    let mut previous = requested.position();
    let mut queries = BTreeSet::new();
    let mut bytes = 0_usize;
    for audit in page.records() {
        if !queries.insert(audit.query().as_str()) {
            return Err(malformed_response(
                "page.records.query",
                "one unmatched listing cannot carry the same query twice",
            ));
        }
        if audit.position() <= previous {
            return Err(malformed_response(
                "page.records.position",
                "positions must be strictly increasing after the requested cursor",
            ));
        }
        previous = audit.position();
        bytes = bytes
            .checked_add(audit.canonical_bytes().len())
            .ok_or_else(|| malformed_response("page.records", "canonical byte count overflowed"))?;
    }
    if page.len() > 1 && bytes > MAX_AUDIT_PAGE_BYTES {
        return Err(malformed_response(
            "page.records",
            "multiple records must fit the query-audit page byte budget",
        ));
    }

    validate_page_cursor(page)
}

fn validate_page_cursor(page: &Page<QueryAudit>) -> Result<(), QueryAuditError> {
    let last = page.records().last().map(Positioned::position);
    if page
        .next_cursor()
        .is_some_and(|cursor| cursor.snapshot().is_some())
    {
        return Err(malformed_response(
            "page.next.snapshot",
            "query-audit cursors have no snapshot binding",
        ));
    }
    if let Some(next) = page.next_cursor()
        && Some(next.position()) != last
    {
        return Err(malformed_response(
            "page.next",
            "must equal the final returned intent position",
        ));
    }
    if !page.is_empty() && page.next_cursor().is_none() {
        return Err(malformed_response(
            "page.next",
            "every nonempty query-audit page names its final record cursor",
        ));
    }
    match page.completeness() {
        PageCompleteness::Truncated if page.is_empty() || page.next_cursor().is_none() => {
            Err(malformed_response(
                "page.next",
                "a truncated page must return a record and its resume cursor",
            ))
        }
        PageCompleteness::Complete if page.is_empty() && page.next_cursor().is_some() => {
            Err(malformed_response(
                "page.next",
                "an empty complete page has no final record to name",
            ))
        }
        PageCompleteness::Complete | PageCompleteness::Truncated => Ok(()),
    }
}

fn receipt(field: impl Into<Option<pb::Receipt>>) -> Result<Receipt, QueryAuditError> {
    let value = field.into().ok_or_else(|| {
        QueryAuditError::from(StateError::Malformed {
            field: "receipt".to_owned(),
            reason: "a successful audit write carries its durable receipt".to_owned(),
        })
    })?;
    Kernel::<Receipt>::try_from(value)
        .map(Kernel::into_inner)
        .map_err(QueryAuditError::from)
}