liminal_server/auth_lys/

grant_stream.rs

//! Explicit startup, stream and console pulls share one serialized executor.

use super::{AtomicDraftStore, DraftError};
use lys_pass::drafts::{ApprovedDraft, Execution};

use super::waiting::durable::{DraftJournal, Outcome};
use super::{ApprovalEvidence, ApprovalTransition, execution_digest, same_request};
use crate::agents::{error::AgentError, runtime::Agents};
use crate::services::lifecycle::site::executor::OutcomeDelivery;
use std::sync::Arc;

/// The native agent owner applies its transition through the approval journal.
pub struct AgentDraftStore {
    pub(super) journal: Arc<DraftJournal>,
    pub(super) agents: Arc<Agents>,
    pub(super) delivery: Arc<dyn OutcomeDelivery>,
    pub(super) projection: Arc<dyn super::AgentProjection>,
}
impl AgentDraftStore {
    /// Uses the registry that was opened over this exact approval journal.
    #[must_use]
    pub fn new(
        owners: &super::waiting::durable::ApprovalAgents,
        delivery: Arc<dyn OutcomeDelivery>,
    ) -> Self {
        Self {
            journal: owners.journal().clone(),
            agents: owners.agents().clone(),
            delivery,
            projection: owners.agents().clone(),
        }
    }
    fn commit_permission(
        &self,
        draft: &ApprovedDraft,
        approval: &ApprovalTransition,
    ) -> Result<Execution, DraftError> {
        let record = self
            .agents
            .registry()
            .record(&draft.target.id)
            .map_err(|_| DraftError::Contract)?;
        if !record.active() {
            return Err(DraftError::Contract);
        }
        crate::auth_lys::prepared::validate_permission_words(&draft.words)
            .map_err(|_| DraftError::Contract)?;
        let receipt = Execution::Executed {
            request_digest: draft.request_digest.clone(),
            receipt_digest: lys_pass::drafts::request_digest(&format!(
                "{}:{}",
                draft.id, draft.request_digest
            )),
        };
        self.journal.update(|snapshot| {
            snapshot.outcomes.insert(
                draft.id.clone(),
                Outcome {
                    draft: Some(draft.clone()),
                    approval: approval.clone(),
                    receipt: receipt.clone(),
                    acknowledged: false,
                    delivered: false,
                },
            );
            Ok(receipt)
        })
    }
}
impl AtomicDraftStore for AgentDraftStore {
    fn receipt(&self, id: &str) -> Result<Option<Execution>, DraftError> {
        self.journal.read(|snapshot| {
            snapshot
                .outcomes
                .get(id)
                .map(|outcome| outcome.receipt.clone())
        })
    }
    fn approval(&self, draft: &ApprovedDraft) -> Result<ApprovalEvidence, DraftError> {
        super::issuer_approval(&self.journal, draft)
    }
    fn commit(
        &mut self,
        draft: &ApprovedDraft,
        approval: &ApprovalTransition,
    ) -> Result<Execution, DraftError> {
        if let Some(receipt) = self.receipt(&draft.id)? {
            return Ok(receipt);
        }
        let prepared = self.journal.read(|snapshot| {
            snapshot
                .prepared
                .values()
                .find(|prepared| prepared.draft.as_deref() == Some(&draft.id))
                .cloned()
        })?;
        let invalid = prepared
            .as_ref()
            .is_none_or(|prepared| !same_request(&prepared.request, draft));
        if !invalid
            && draft.target.kind == "liminal.agent"
            && draft.target.action == "agent_permission"
        {
            return self.commit_permission(draft, approval);
        }
        let receipt = if invalid {
            Execution::RefusedOnExecution {
                request_digest: draft.request_digest.clone(),
                refusal: "draft_preparation_mismatch".into(),
                reason: "approved words do not match a durable prepared operation".into(),
            }
        } else {
            if let Err(error) = self.journal.begin(draft, approval) {
                if !matches!(error, DraftError::Contract | DraftError::Bound) {
                    return Err(error);
                }
                let refusal = Execution::RefusedOnExecution {
                    request_digest: draft.request_digest.clone(),
                    refusal: error.to_string(),
                    reason: "prepared operation no longer validates within its target bounds"
                        .into(),
                };
                return self.journal.update(|snapshot| {
                    snapshot.outcomes.insert(
                        draft.id.clone(),
                        Outcome {
                            draft: Some(draft.clone()),
                            approval: approval.clone(),
                            receipt: refusal.clone(),
                            acknowledged: false,
                            delivered: false,
                        },
                    );
                    Ok(refusal)
                });
            }
            let result = self
                .agents
                .registry()
                .execute_approved_at(draft, approval.executed_at_ms);
            self.journal.finish()?;
            match result {
                Ok(receipt) => {
                    if self.receipt(&draft.id)?.as_ref() != Some(&receipt) {
                        return Err(DraftError::Store);
                    }
                    return Ok(receipt);
                }
                Err(AgentError::Store(_) | AgentError::Uncertain | AgentError::Delivery(_)) => {
                    return Err(DraftError::Store);
                }
                Err(error) => Execution::RefusedOnExecution {
                    request_digest: draft.request_digest.clone(),
                    refusal: agent_refusal(&error).into(),
                    reason: error.to_string(),
                },
            }
        };
        self.journal.update(|snapshot| {
            snapshot.outcomes.insert(
                draft.id.clone(),
                Outcome {
                    draft: Some(draft.clone()),
                    approval: approval.clone(),
                    receipt: receipt.clone(),
                    acknowledged: false,
                    delivered: false,
                },
            );
            Ok(receipt)
        })
    }
    fn acknowledged(&self, id: &str) -> Result<bool, DraftError> {
        self.journal.read(|snapshot| {
            snapshot
                .outcomes
                .get(id)
                .is_some_and(|outcome| outcome.acknowledged)
        })
    }
    fn acknowledge(&mut self, id: &str, receipt: &Execution) -> Result<(), DraftError> {
        self.journal.update(|snapshot| {
            let outcome = snapshot
                .outcomes
                .get_mut(id)
                .ok_or(DraftError::ReceiptMismatch)?;
            if &outcome.receipt != receipt {
                return Err(DraftError::ReceiptMismatch);
            }
            outcome.acknowledged = true;
            Ok(())
        })
    }
    fn pending_acknowledgments(
        &self,
        limit: std::num::NonZeroUsize,
    ) -> Result<Vec<ApprovedDraft>, DraftError> {
        self.journal.read(|snapshot| {
            snapshot
                .outcomes
                .values()
                .filter(|outcome| !outcome.acknowledged)
                .filter_map(|outcome| outcome.draft.clone())
                .take(limit.get())
                .collect()
        })
    }
    fn deliver_outcome(
        &mut self,
        draft: &ApprovedDraft,
        receipt: &Execution,
    ) -> Result<(), DraftError> {
        let outcome = self
            .journal
            .read(|snapshot| snapshot.outcomes.get(&draft.id).cloned())?
            .ok_or(DraftError::ReceiptMismatch)?;
        if &outcome.receipt != receipt || execution_digest(receipt) != draft.request_digest {
            return Err(DraftError::ReceiptMismatch);
        }
        if outcome.delivered {
            return Ok(());
        }
        let holder = self.journal.read(|snapshot| {
            snapshot
                .prepared
                .values()
                .find(|prepared| prepared.draft.as_deref() == Some(&draft.id))
                .map(|prepared| prepared.holder.clone())
        })?;
        if let Some(holder) = holder {
            self.delivery.deliver(&holder, draft, receipt)?;
        }
        if matches!(receipt, Execution::Executed { .. }) {
            self.projection.apply(draft)?;
        }
        self.journal.update(|snapshot| {
            let outcome = snapshot
                .outcomes
                .get_mut(&draft.id)
                .ok_or(DraftError::Store)?;
            outcome.draft = Some(draft.clone());
            outcome.delivered = true;
            Ok(())
        })
    }
}

mod journal_store {
    use super::super::waiting::durable::DraftJournal;
    use crate::agents::registry::REGISTRY_STREAM;
    use crate::auth_lys::live::drafts::DraftError;
    use liminal::durability::{
        DurabilityError, DurableStore, RoutedBatchEntry, RoutedBatchItem, ScanCursor, ScanPage,
        StoredEntry,
    };
    use std::sync::Arc;
    #[async_trait::async_trait]
    impl DurableStore for DraftJournal {
        fn max_request_bytes(&self) -> Option<usize> {
            self.store.max_request_bytes()
        }
        fn reclaim_spine(&self) -> Option<Arc<haematite::EventStore>> {
            self.store.reclaim_spine()
        }
        fn store_counts(&self) -> Option<liminal::durability::StoreCountsHandle> {
            self.store.store_counts()
        }
        async fn apply_routed_batch(
            &self,
            route: &str,
            items: Vec<RoutedBatchItem>,
        ) -> Result<(), DurabilityError> {
            self.store.apply_routed_batch(route, items).await
        }
        async fn apply_routed_entries(
            &self,
            route: &str,
            entries: Vec<RoutedBatchEntry>,
        ) -> Result<(), DurabilityError> {
            self.store.apply_routed_entries(route, entries).await
        }
        async fn get_routed(
            &self,
            route: &str,
            key: &[u8],
        ) -> Result<Option<Vec<u8>>, DurabilityError> {
            self.store.get_routed(route, key).await
        }
        async fn append(
            &self,
            stream: &str,
            payload: Vec<u8>,
            expected: u64,
        ) -> Result<u64, DurabilityError> {
            if stream == REGISTRY_STREAM {
                self.append_native(vec![payload], expected)
                    .map_err(|error| Self::storage(&error))?;
                Ok(expected)
            } else {
                self.store.append(stream, payload, expected).await
            }
        }
        async fn append_all(
            &self,
            stream: &str,
            payloads: Vec<Vec<u8>>,
            expected: u64,
        ) -> Result<u64, DurabilityError> {
            if stream == REGISTRY_STREAM {
                self.append_native(payloads, expected)
                    .map_err(|error| Self::storage(&error))
            } else {
                self.store.append_all(stream, payloads, expected).await
            }
        }
        async fn read_from(
            &self,
            stream: &str,
            offset: u64,
            limit: usize,
        ) -> Result<Vec<StoredEntry>, DurabilityError> {
            if stream != REGISTRY_STREAM {
                return self.store.read_from(stream, offset, limit).await;
            }
            self.read(|snapshot| {
                let offset = usize::try_from(offset).map_err(|_| DraftError::Bound)?;
                snapshot
                    .rows
                    .iter()
                    .enumerate()
                    .skip(offset)
                    .take(limit)
                    .map(|(sequence, row)| {
                        Ok(StoredEntry {
                            payload: serde_json::to_vec(&row.target)
                                .map_err(|_| DraftError::Store)?,
                            sequence: u64::try_from(sequence).map_err(|_| DraftError::Bound)?,
                            timestamp: row.timestamp,
                        })
                    })
                    .collect::<Result<Vec<_>, DraftError>>()
            })
            .map_err(|error| Self::storage(&error))?
            .map_err(|error| Self::storage(&error))
        }
        async fn cas(&self, key: &str, old: u64, new: u64) -> Result<(), DurabilityError> {
            self.store.cas(key, old, new).await
        }
        async fn read_stream_next_seq(&self, stream: &str) -> Result<Option<u64>, DurabilityError> {
            if stream != REGISTRY_STREAM {
                return self.store.read_stream_next_seq(stream).await;
            }
            self.read(|snapshot| {
                if snapshot.rows.is_empty() {
                    Ok(None)
                } else {
                    u64::try_from(snapshot.rows.len())
                        .map(Some)
                        .map_err(|_| DraftError::Bound)
                }
            })
            .map_err(|error| Self::storage(&error))?
            .map_err(|error| Self::storage(&error))
        }
        async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
            self.store.read_value(key).await
        }
        async fn scan_page(
            &self,
            prefix: &str,
            cursor: Option<ScanCursor>,
            limit: usize,
        ) -> Result<ScanPage, DurabilityError> {
            if REGISTRY_STREAM.starts_with(prefix) || prefix.starts_with(REGISTRY_STREAM) {
                return Err(DurabilityError::ConfigError(
                    "agent approval history requires its exact stream reader".into(),
                ));
            }
            self.store.scan_page(prefix, cursor, limit).await
        }
        async fn flush(&self) -> Result<(), DurabilityError> {
            self.flush_pending()
        }
        async fn truncate_before(&self, stream: &str, before: u64) -> Result<u64, DurabilityError> {
            if stream == REGISTRY_STREAM {
                return Err(DurabilityError::TruncateUnsupported {
                    store: "DraftJournal",
                    stream_key: stream.into(),
                    before_seq: before,
                });
            }
            self.store.truncate_before(stream, before).await
        }
        async fn scan_routed(
            &self,
            route: &str,
            from: &[u8],
            to: &[u8],
        ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
            self.store.scan_routed(route, from, to).await
        }
        async fn remove_routed_range(
            &self,
            route: &str,
            from: &[u8],
            to: &[u8],
        ) -> Result<u64, DurabilityError> {
            self.store.remove_routed_range(route, from, to).await
        }
    }
}

impl super::TargetDraftStore {
    pub(super) fn receipt_owner(&self, id: &str) -> Result<bool, DraftError> {
        let agent = self.agents.receipt(id)?.is_some();
        let site = self.sites.receipt(id)?.is_some();
        if agent == site {
            return Err(DraftError::ReceiptMismatch);
        }
        Ok(agent)
    }
}

const fn agent_refusal(error: &AgentError) -> &'static str {
    match error {
        AgentError::Unknown(_) => "unknown_agent",
        AgentError::NameTaken(_) => "agent_name_taken",
        AgentError::Transition(_) => "agent_transition_refused",
        AgentError::Invalid(_) => "agent_words_invalid",
        AgentError::Bound { .. } => "agent_bound_exceeded",
        AgentError::DraftMismatch => "agent_draft_digest_mismatch",
        _ => "agent_execution_refused",
    }
}

pub struct Kept {
    pub(crate) snapshot: super::waiting::durable::Snapshot,
    pub(crate) hash: Option<haematite::tree::Hash>,
    pub(crate) uncertain: bool,
    pub(crate) active: Option<(ApprovedDraft, ApprovalTransition)>,
    pub(crate) pending: Option<(super::waiting::durable::Snapshot, haematite::tree::Hash)>,
}

#[cfg(test)]
#[path = "grant_stream_tests.rs"]
mod tests;