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;
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 {
#[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;