liminal_server/agents/

waiting.rs

//! Retained waiting and outcome messages, changed in the agent's own store.

use std::collections::BTreeMap;
use std::num::NonZeroUsize;

use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

#[path = "waiting_state.rs"]
pub mod durable;

/// Bounds on one agent's retained approval history.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WaitingLimits {
    /// Bounds pending drafts and resolved ids together, preserving deduplication
    /// without evicting evidence. A full record refuses another held act.
    pub drafts: NonZeroUsize,
    /// Bounds exact approver words; an oversized draft is refused, never cut.
    pub words_bytes: NonZeroUsize,
    /// Bounds identifiers used to form retained-state and inbox subjects.
    pub identifier_bytes: NonZeroUsize,
}

#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct HeldDraft {
    draft: String,
    words: String,
    since: u64,
}

#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct ResolvedDraft {
    digest: String,
    outcome: DraftOutcome,
}

/// The same outcome is retained on state and published on the inbox.
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DraftOutcome {
    /// The immutable draft id.
    pub draft: String,
    /// Executed or refused; never an authority decision made here.
    pub outcome: OutcomeKind,
    /// The named execution refusal, if any.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub refusal: Option<String>,
    /// The refusing authority's exact reason, unchanged.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OutcomeKind {
    Executed,
    Refused,
}

impl DraftOutcome {
    /// An execution already committed by the target store.
    #[must_use]
    pub fn executed(draft: &str) -> Self {
        Self {
            draft: draft.to_owned(),
            outcome: OutcomeKind::Executed,
            refusal: None,
            reason: None,
        }
    }

    /// A refusal already committed by the authority or target store.
    #[must_use]
    pub fn refused(draft: &str, refusal: &str, reason: &str) -> Self {
        Self {
            draft: draft.to_owned(),
            outcome: OutcomeKind::Refused,
            refusal: Some(refusal.to_owned()),
            reason: Some(reason.to_owned()),
        }
    }
}

#[derive(Clone, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WaitingRecord {
    pending: Vec<HeldDraft>,
    resolved: BTreeMap<String, ResolvedDraft>,
    last: Option<DraftOutcome>,
    gone: bool,
}

impl std::fmt::Debug for WaitingRecord {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("WaitingRecord")
            .finish_non_exhaustive()
    }
}

/// Named failures contain no held words or credential bytes.
#[derive(Debug, thiserror::Error)]
pub enum WaitingError {
    /// An input does not fit the configured retained record.
    #[error("agent_waiting_bound")]
    Bound,
    /// The same draft id was presented with different words or outcome.
    #[error("agent_waiting_draft_conflict")]
    Conflict,
    /// No held draft has this id.
    #[error("agent_waiting_draft_unknown")]
    Unknown,
    /// State could not be encoded or durably committed.
    #[error("agent_waiting_store_failed")]
    Store,
}

impl WaitingRecord {
    /// Retains exact words once, with the earliest held instant.
    ///
    /// # Errors
    /// Invalid ids, empty or oversized words, capacity, or conflicting replay.
    pub fn hold(
        &mut self,
        draft: &str,
        words: &str,
        since: u64,
        limits: WaitingLimits,
    ) -> Result<bool, WaitingError> {
        if draft.is_empty()
            || draft.len() > limits.identifier_bytes.get()
            || draft.chars().any(char::is_control)
            || words.is_empty()
            || words.len() > limits.words_bytes.get()
        {
            return Err(WaitingError::Bound);
        }
        if let Some(previous) = self.resolved.get(draft) {
            return if previous.digest == lys_pass::drafts::request_digest(words) {
                Ok(false)
            } else {
                Err(WaitingError::Conflict)
            };
        }
        if let Some(held) = self.pending.iter().find(|held| held.draft == draft) {
            return if held.words == words {
                Ok(false)
            } else {
                Err(WaitingError::Conflict)
            };
        }
        if self.pending.len().saturating_add(self.resolved.len()) >= limits.drafts.get() {
            return Err(WaitingError::Bound);
        }
        self.pending.push(HeldDraft {
            draft: draft.to_owned(),
            words: words.to_owned(),
            since,
        });
        self.pending
            .sort_by(|left, right| (left.since, &left.draft).cmp(&(right.since, &right.draft)));
        Ok(true)
    }

    /// Completes one held draft and returns its inbox message once.
    /// The caller commits this record, retained state and inbox together.
    ///
    /// # Errors
    /// Unknown or conflicting draft, invalid outcome, or missing provenance.
    pub fn resolve(
        &mut self,
        outcome: &DraftOutcome,
        exchange: &str,
        gone: bool,
        limits: WaitingLimits,
    ) -> Result<Option<Value>, WaitingError> {
        if exchange.len() > limits.identifier_bytes.get()
            || outcome.draft.len() > limits.identifier_bytes.get()
            || outcome
                .refusal
                .as_ref()
                .is_some_and(|name| name.len() > limits.identifier_bytes.get())
            || outcome
                .reason
                .as_ref()
                .is_some_and(|reason| reason.len() > limits.words_bytes.get())
        {
            return Err(WaitingError::Bound);
        }
        if exchange.is_empty()
            || exchange.chars().any(char::is_control)
            || outcome.draft.is_empty()
            || outcome.draft.chars().any(char::is_control)
            || outcome
                .refusal
                .as_ref()
                .is_some_and(|name| name.chars().any(char::is_control))
            || matches!(outcome.outcome, OutcomeKind::Refused)
                && (outcome.refusal.as_deref().is_none_or(str::is_empty)
                    || outcome.reason.as_deref().is_none_or(str::is_empty))
            || matches!(outcome.outcome, OutcomeKind::Executed)
                && (outcome.refusal.is_some() || outcome.reason.is_some())
        {
            return Err(WaitingError::Conflict);
        }
        if let Some(previous) = self.resolved.get(&outcome.draft) {
            return if &previous.outcome == outcome {
                Ok(None)
            } else {
                Err(WaitingError::Conflict)
            };
        }
        let position = self
            .pending
            .iter()
            .position(|held| held.draft == outcome.draft)
            .ok_or(WaitingError::Unknown)?;
        let mut inbox = serde_json::to_value(outcome).map_err(|_| WaitingError::Store)?;
        inbox["meta"] = json!({"from":exchange});
        let held = self.pending.remove(position);
        self.resolved.insert(
            outcome.draft.clone(),
            ResolvedDraft {
                digest: lys_pass::drafts::request_digest(&held.words),
                outcome: outcome.clone(),
            },
        );
        self.last = Some(outcome.clone());
        self.gone = gone;
        Ok(Some(inbox))
    }

    /// The oldest pending draft and total count, or the last terminal outcome.
    ///
    /// # Errors
    /// The terminal outcome could not be encoded.
    pub fn snapshot(&self) -> Result<Value, WaitingError> {
        if let Some(held) = self.pending.first() {
            return Ok(json!({"state":"waiting", "draft":held.draft,
                "words":held.words, "since":held.since, "count":self.pending.len(), "last_outcome":self.last}));
        }
        let mut value = match &self.last {
            Some(last) => serde_json::to_value(last).map_err(|_| WaitingError::Store)?,
            None => json!({}),
        };
        value["state"] = json!(if self.gone { "gone" } else { "present" });
        Ok(value)
    }
}

use crate::agents::runtime::Agents;
use crate::auth_lys::live::drafts::DraftError;
use durable::DraftJournal;
use liminal::channel::ChannelHandle;
use lys_pass::drafts::{ApprovedDraft, Execution};
use std::sync::{Arc, Mutex};

#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Delivery {
    agent: String,
    from: Vec<u64>,
    message: Value,
    state: Value,
    delivered: bool,
}
/// Publishes the durable outbox to the verified holder's server-resolved inbox.
pub struct AgentDelivery {
    agents: Arc<Agents>,
    journal: Arc<DraftJournal>,
    inboxes: Mutex<BTreeMap<String, ChannelHandle>>,
    exchange: String,
    limits: WaitingLimits,
    serial: Mutex<()>,
}
impl AgentDelivery {
    /// Exact native durable inboxes share the configured retained-history population bound.
    ///
    /// # Errors
    /// Invalid station, mismatched/non-durable inbox or configured capacity.
    pub fn new(
        owners: &durable::ApprovalAgents,
        inboxes: BTreeMap<String, ChannelHandle>,
        exchange: String,
        limits: WaitingLimits,
    ) -> Result<Arc<Self>, DraftError> {
        if inboxes.len() > limits.drafts.get()
            || exchange.is_empty()
            || exchange.len() > limits.identifier_bytes.get()
            || exchange.chars().any(char::is_control)
        {
            return Err(DraftError::Bound);
        }
        for (name, inbox) in &inboxes {
            owners
                .agents()
                .registry()
                .record(name)
                .map_err(|_| DraftError::Contract)?;
            if inbox.config().name != format!("agents.{name}.inbox") {
                return Err(DraftError::Contract);
            }
            inbox
                .durable_positions()
                .map_err(|_| DraftError::Contract)?;
        }
        Ok(Arc::new(Self {
            agents: owners.agents().clone(),
            journal: owners.journal().clone(),
            inboxes: Mutex::new(inboxes),
            exchange,
            limits,
            serial: Mutex::new(()),
        }))
    }
    fn message(
        &self,
        agent: &str,
        draft: &ApprovedDraft,
        receipt: &Execution,
        from: Vec<u64>,
    ) -> Result<Delivery, DraftError> {
        let outcome = match receipt {
            Execution::Executed { .. } => DraftOutcome::executed(&draft.id),
            Execution::RefusedOnExecution {
                refusal, reason, ..
            } => DraftOutcome::refused(&draft.id, refusal, reason),
        };
        let gone = self
            .agents
            .presence()
            .connection(agent)
            .map_err(|_| DraftError::Store)?
            .is_none();
        self.journal.update(|snapshot| {
            if let Some(old) = snapshot.deliveries.get(&draft.id) {
                return Ok(old.clone());
            }
            let waiting = snapshot
                .waiting
                .get_mut(agent)
                .ok_or(DraftError::Contract)?;
            let message = waiting
                .resolve(&outcome, &self.exchange, gone, self.limits)
                .map_err(|_| DraftError::Contract)?
                .ok_or(DraftError::Contract)?;
            let delivery = Delivery {
                agent: agent.into(),
                from,
                message,
                state: waiting.snapshot().map_err(|_| DraftError::Store)?,
                delivered: false,
            };
            snapshot
                .deliveries
                .insert(draft.id.clone(), delivery.clone());
            Ok(delivery)
        })
    }
    /// Binds a newly registered native inbox without rebuilding the outbox owner.
    ///
    /// # Errors
    /// Unknown station, wrong/non-durable channel, unavailable index or capacity.
    pub fn bind_inbox(&self, name: &str, inbox: ChannelHandle) -> Result<(), DraftError> {
        self.agents
            .registry()
            .record(name)
            .map_err(|_| DraftError::Contract)?;
        if inbox.config().name != format!("agents.{name}.inbox") {
            return Err(DraftError::Contract);
        }
        inbox
            .durable_positions()
            .map_err(|_| DraftError::Contract)?;
        let mut inboxes = self.inboxes.lock().map_err(|_| DraftError::Store)?;
        if inboxes.contains_key(name) {
            return Ok(());
        }
        if inboxes.len() >= self.limits.drafts.get() {
            return Err(DraftError::Bound);
        }
        inboxes.insert(name.into(), inbox);
        drop(inboxes);
        Ok(())
    }
}
#[path = "waiting_outcome.rs"]
mod outcome;

#[cfg(test)]
#[path = "waiting_tests.rs"]
pub(crate) mod tests;

impl std::fmt::Debug for AgentDelivery {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AgentDelivery").finish_non_exhaustive()
    }
}