liminal_server/services/site/

executor.rs

//! The construction target for the approved-draft executor and durable outbox.

use lys_pass::drafts::{ApprovedDraft, DraftRequest, Execution};
use serde::{Deserialize, Serialize};
use std::collections::BTreeSet;
use std::num::NonZeroUsize;
use std::sync::Arc;

use super::super::Lifecycle;
use super::{SiteAct, SiteError};
use crate::auth_lys::live::drafts::{
    ApprovalEvidence, ApprovalTransition, AtomicDraftStore, DraftError,
};

/// Immutable words kept only in their owning durable target store.
#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(in super::super) struct Prepared {
    pub draft: ApprovedDraft,
    pub act: SiteAct,
    pub at_ms: u64,
}

/// The outcome and notification intent share the target's durable operation.
#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(in super::super) struct Outcome {
    pub receipt: Execution,
    pub approval: ApprovalTransition,
    pub acknowledged: bool,
    pub delivered: bool,
}

/// The approval owner authenticates the actual people and immutable words.
pub trait ApprovalReader: Send + Sync {
    /// Reads evidence for this exact draft from Lys's authenticated record.
    ///
    /// # Errors
    /// Missing evidence or an issuer/storage refusal.
    fn read(&self, draft: &ApprovedDraft) -> Result<ApprovalEvidence, DraftError>;
}

/// The agent owner delivers by immutable operation id through its state writer.
pub trait OutcomeDelivery: Send + Sync {
    /// Atomically clears waiting and puts one outcome in the bound agent inbox.
    ///
    /// # Errors
    /// Missing agent binding, storage or publication failure; no guessed identity.
    fn deliver(
        &self,
        holder: &str,
        draft: &ApprovedDraft,
        receipt: &Execution,
    ) -> Result<(), DraftError>;
}

/// No process, timer or authority is created by this target adapter.
pub struct SiteDraftStore {
    lifecycle: Arc<Lifecycle>,
    approvals: Arc<dyn ApprovalReader>,
    delivery: Arc<dyn OutcomeDelivery>,
    at_ms: u64,
}

impl SiteDraftStore {
    /// Joins the existing approval reader and agent outcome owner.
    #[must_use]
    pub fn new(
        lifecycle: Arc<Lifecycle>,
        approvals: Arc<dyn ApprovalReader>,
        delivery: Arc<dyn OutcomeDelivery>,
        at_ms: u64,
    ) -> Self {
        Self {
            lifecycle,
            approvals,
            delivery,
            at_ms,
        }
    }

    fn validate_approval(
        &self,
        draft: &ApprovedDraft,
        transition: &ApprovalTransition,
    ) -> Result<(), DraftError> {
        let evidence = self.approvals.read(draft)?;
        let needed = match evidence.mode {
            crate::auth_lys::GrantMode::ByDraft => 1,
            crate::auth_lys::GrantMode::ByTwo => 2,
            crate::auth_lys::GrantMode::Outright => return Err(DraftError::ApprovalUnavailable),
        };
        let limits = &self
            .lifecycle
            .site_bindings()
            .map_err(storage_error)?
            .limits;
        let mut people = BTreeSet::new();
        if evidence.draft != draft.id
            || evidence.digest != draft.request_digest
            || evidence.approved_by != transition.approved_by
            || evidence.approved_by.len() != needed
            || evidence.approved_by.len() > limits.approvers.get()
            || evidence.approved_by.iter().any(|person| {
                person == "lys"
                    || person.is_empty()
                    || person.len() > limits.text_bytes.get()
                    || person.chars().any(char::is_control)
                    || !people.insert(person)
            })
        {
            return Err(DraftError::ApprovalUnavailable);
        }
        Ok(())
    }

    /// Reads the target receipt through the approval owner's atomic interface.
    ///
    /// # Errors
    /// The durable target ledger cannot be read.
    pub fn receipt(&self, id: &str) -> Result<Option<Execution>, DraftError> {
        AtomicDraftStore::receipt(self, id)
    }

    /// The operation clock is observed by the pulling owner, never the request.
    pub const fn observe_execution_time(&mut self, at_ms: u64) {
        self.at_ms = at_ms;
    }

    /// Validates every immutable field even on receipt replay, then delegates
    /// execution and durable deduplication to Lys's shared executor contract.
    ///
    /// # Errors
    /// A conflicting target, grant, digest, words or execution failure.
    pub fn execute_approved(&mut self, draft: &ApprovedDraft) -> Result<Execution, DraftError> {
        {
            let registry = self.lifecycle.lock().map_err(|_| DraftError::Store)?;
            let prepared = registry
                .construction
                .drafts
                .get(&draft.id)
                .ok_or(DraftError::Contract)?;
            if !same_draft(&prepared.draft, draft) {
                return Err(DraftError::ReceiptMismatch);
            }
            drop(registry);
        }
        let receipt = lys_pass::drafts::execute_once(draft, self).map_err(|error| match error {
            lys_pass::drafts::ExecuteError::DigestMismatch => DraftError::ReceiptMismatch,
            lys_pass::drafts::ExecuteError::Product(error) => error,
            lys_pass::drafts::ExecuteError::Contract(_) => DraftError::Contract,
        })?;
        self.lifecycle.refresh_sites().map_err(storage_error)?;
        Ok(receipt)
    }
}

impl Lifecycle {
    /// Records the exact prepared request after Lys durably holds it. Identity
    /// is checked again against the original verified pass, never request fields.
    ///
    /// # Errors
    /// Any immutable mismatch, invalid boundary or durable capacity failure.
    pub fn keep_held_site(
        &self,
        id: &str,
        request: &DraftRequest,
        act: &SiteAct,
        pass: &lys_pass::VerifiedPass,
        mode: crate::auth_lys::GrantMode,
        at_ms: u64,
    ) -> Result<(), SiteError> {
        let bindings = self.site_bindings()?;
        let bound = bindings.limits.text_bytes.get();
        if [
            id,
            &request.grant,
            &act.target,
            &act.holder,
            &act.responsible,
        ]
        .iter()
        .any(|value| value.is_empty() || value.len() > bound || value.chars().any(char::is_control))
        {
            return Err(SiteError::Invalid("held metadata"));
        }
        let verified = SiteAct::prepare(act.target.clone(), act.change.clone(), pass)?;
        let exact = serde_json::to_string(&verified).map_err(|_| SiteError::Store)?;
        if exact != request.words
            || request.words.len() > bindings.limits.request_bytes.get()
            || request.request_digest != lys_pass::drafts::request_digest(&exact)
            || request.target.kind != act.change.kind()
            || request.target.id != act.permission_id(self.network_id())
            || request.target.action != act.change.action()
        {
            return Err(SiteError::Invalid("held immutable request"));
        }
        // Held, so no approver yet; the mode is the one its preparation recorded.
        let mode = match mode {
            crate::auth_lys::GrantMode::ByDraft => lys_pass::Mode::ByDraft,
            crate::auth_lys::GrantMode::ByTwo => lys_pass::Mode::ByTwo,
            crate::auth_lys::GrantMode::Outright => {
                return Err(SiteError::Invalid("held approval mode"));
            }
        };
        let draft = ApprovedDraft {
            id: id.into(),
            app: "liminal".into(),
            grant: request.grant.clone(),
            target: request.target.clone(),
            request_digest: request.request_digest.clone(),
            approved_by: Vec::new(),
            mode,
            words: exact,
        };
        let mut registry = self.lock()?;
        if let Some(old) = registry.construction.drafts.get(id) {
            return if same_draft(&old.draft, &draft) {
                Ok(())
            } else {
                Err(SiteError::Invalid("held draft conflict"))
            };
        }
        if registry.construction.drafts.len() >= bindings.limits.drafts.get() {
            return Err(SiteError::Bound("drafts"));
        }
        let mut snapshot = Self::snapshot(&registry);
        snapshot.construction.drafts.insert(
            id.into(),
            Prepared {
                draft,
                act: verified,
                at_ms,
            },
        );
        super::storage::commit(self, &mut registry, &snapshot)?;
        drop(registry);
        Ok(())
    }
}

fn same_draft(left: &ApprovedDraft, right: &ApprovedDraft) -> bool {
    left.id == right.id
        && left.app == right.app
        && left.grant == right.grant
        && left.target.kind == right.target.kind
        && left.target.id == right.target.id
        && left.target.action == right.target.action
        && left.request_digest == right.request_digest
        && left.words == right.words
}

fn storage_error(_: SiteError) -> DraftError {
    DraftError::Store
}

impl AtomicDraftStore for SiteDraftStore {
    fn receipt(&self, draft: &str) -> Result<Option<Execution>, DraftError> {
        Ok(self
            .lifecycle
            .lock()
            .map_err(|_| DraftError::Store)?
            .construction
            .outcomes
            .get(draft)
            .map(|outcome| outcome.receipt.clone()))
    }
    fn approval(&self, draft: &ApprovedDraft) -> Result<ApprovalEvidence, DraftError> {
        self.approvals.read(draft)
    }
    fn commit(
        &mut self,
        draft: &ApprovedDraft,
        transition: &ApprovalTransition,
    ) -> Result<Execution, DraftError> {
        self.validate_approval(draft, transition)?;
        let mut registry = self.lifecycle.lock().map_err(|_| DraftError::Store)?;
        if let Some(outcome) = registry.construction.outcomes.get(&draft.id) {
            let prepared = registry
                .construction
                .drafts
                .get(&draft.id)
                .ok_or(DraftError::Contract)?;
            if !same_draft(&prepared.draft, draft) {
                return Err(DraftError::ReceiptMismatch);
            }
            return Ok(outcome.receipt.clone());
        }
        let prepared = registry
            .construction
            .drafts
            .get(&draft.id)
            .cloned()
            .ok_or(DraftError::Contract)?;
        if !same_draft(&prepared.draft, draft)
            || transition.draft != draft.id
            || transition.digest != draft.request_digest
            || transition.approved_by.is_empty()
        {
            return Err(DraftError::ReceiptMismatch);
        }
        let mut snapshot = Lifecycle::snapshot(&registry);
        let applied = self.lifecycle.apply_site_act(
            &mut snapshot,
            &prepared.act,
            transition,
            prepared.at_ms,
            &registry.trials,
        );
        let receipt = match applied {
            Ok(()) => {
                let bytes = serde_json::to_string(&(
                    transition,
                    &snapshot.records,
                    &snapshot.construction.templates,
                ))
                .map_err(|_| DraftError::Store)?;
                Execution::Executed {
                    request_digest: draft.request_digest.clone(),
                    receipt_digest: lys_pass::drafts::request_digest(&bytes),
                }
            }
            Err(SiteError::Pass(name)) => return Err(DraftError::Lys(name)),
            Err(SiteError::Store | SiteError::Lifecycle(_)) => {
                return Err(DraftError::Store);
            }
            Err(error) => {
                snapshot = Lifecycle::snapshot(&registry);
                Execution::RefusedOnExecution {
                    request_digest: draft.request_digest.clone(),
                    refusal: "site_execution_refused".into(),
                    reason: error.to_string(),
                }
            }
        };
        if matches!(receipt, Execution::Executed { .. }) {
            for record in snapshot.records.values_mut() {
                if let Some(approval) = record
                    .approvals
                    .last_mut()
                    .filter(|approval| approval.draft == draft.id)
                {
                    approval.receipt = Some(receipt.clone());
                }
            }
            for template in snapshot.construction.templates.values_mut() {
                if let Some(approval) = template
                    .transitions
                    .last_mut()
                    .filter(|approval| approval.draft == draft.id)
                {
                    approval.receipt = Some(receipt.clone());
                }
            }
        }
        snapshot.construction.outcomes.insert(
            draft.id.clone(),
            Outcome {
                receipt: receipt.clone(),
                approval: transition.clone(),
                acknowledged: false,
                delivered: false,
            },
        );
        let result = super::storage::commit(&self.lifecycle, &mut registry, &snapshot)
            .map_err(|_| DraftError::Store);
        drop(registry);
        result?;
        self.lifecycle.refresh_sites().map_err(storage_error)?;
        Ok(receipt)
    }
    fn acknowledged(&self, draft: &str) -> Result<bool, DraftError> {
        Ok(self
            .lifecycle
            .lock()
            .map_err(|_| DraftError::Store)?
            .construction
            .outcomes
            .get(draft)
            .is_some_and(|outcome| outcome.acknowledged))
    }
    fn acknowledge(&mut self, draft: &str, receipt: &Execution) -> Result<(), DraftError> {
        let mut registry = self.lifecycle.lock().map_err(|_| DraftError::Store)?;
        let mut snapshot = Lifecycle::snapshot(&registry);
        let outcome = snapshot
            .construction
            .outcomes
            .get_mut(draft)
            .ok_or(DraftError::ReceiptMismatch)?;
        if &outcome.receipt != receipt {
            return Err(DraftError::ReceiptMismatch);
        }
        if outcome.acknowledged {
            return Ok(());
        }
        outcome.acknowledged = true;
        let result = super::storage::commit(&self.lifecycle, &mut registry, &snapshot)
            .map_err(|_| DraftError::Store);
        drop(registry);
        result
    }
    fn pending_acknowledgments(
        &self,
        limit: NonZeroUsize,
    ) -> Result<Vec<ApprovedDraft>, DraftError> {
        let registry = self.lifecycle.lock().map_err(|_| DraftError::Store)?;
        registry
            .construction
            .outcomes
            .iter()
            .filter(|(_, outcome)| !outcome.acknowledged)
            .take(limit.get())
            .map(|(id, _)| {
                registry
                    .construction
                    .drafts
                    .get(id)
                    .map(|prepared| prepared.draft.clone())
                    .ok_or(DraftError::Store)
            })
            .collect()
    }
    fn deliver_outcome(
        &mut self,
        draft: &ApprovedDraft,
        receipt: &Execution,
    ) -> Result<(), DraftError> {
        let (holder, delivered) = {
            let registry = self.lifecycle.lock().map_err(|_| DraftError::Store)?;
            let outcome = registry
                .construction
                .outcomes
                .get(&draft.id)
                .ok_or(DraftError::ReceiptMismatch)?;
            let prepared = registry
                .construction
                .drafts
                .get(&draft.id)
                .ok_or(DraftError::ReceiptMismatch)?;
            if &outcome.receipt != receipt || !same_draft(&prepared.draft, draft) {
                return Err(DraftError::ReceiptMismatch);
            }
            let result = (prepared.act.holder.clone(), outcome.delivered);
            drop(registry);
            result
        };
        if delivered {
            return Ok(());
        }
        self.delivery.deliver(&holder, draft, receipt)?;
        let mut registry = self.lifecycle.lock().map_err(|_| DraftError::Store)?;
        let mut snapshot = Lifecycle::snapshot(&registry);
        snapshot
            .construction
            .outcomes
            .get_mut(&draft.id)
            .ok_or(DraftError::Store)?
            .delivered = true;
        let result = super::storage::commit(&self.lifecycle, &mut registry, &snapshot)
            .map_err(|_| DraftError::Store);
        drop(registry);
        result
    }
}

#[path = "executor_run.rs"]
mod run;