liminal_server/agents/

waiting_state.rs

//! Atomic agent target journal and the prepared-operation ledger.

use super::WaitingRecord;
use crate::agents::registry::REGISTRY_STREAM;
use crate::auth_lys::live::drafts::{ApprovalTransition, DraftError, execution_digest};
use haematite::tree::Hash;
use liminal::durability::{
    DurabilityError, DurableStore, RoutedBatchItem, ScanCursor, bridge::block_on,
};
use lys_pass::drafts::{ApprovedDraft, DraftRequest, Execution};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex, MutexGuard};

const ROUTE: &str = "liminal/approvals/v2";
const KEY: &[u8] = b"snapshot";

pub use crate::auth_lys::live::drafts::JournalLimits;

#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Prepared {
    pub holder: String,
    pub agent: Option<String>,
    pub request: DraftRequest,
    pub at_ms: u64,
    pub mode: crate::auth_lys::GrantMode,
    pub draft: Option<String>,
}
#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Outcome {
    pub draft: Option<ApprovedDraft>,
    pub approval: ApprovalTransition,
    pub receipt: Execution,
    pub acknowledged: bool,
    pub delivered: bool,
}
#[derive(Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TargetRow {
    pub(crate) timestamp: u64,
    pub(crate) target: Value,
    approval: Option<ApprovalTransition>,
}
#[derive(Clone, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Snapshot {
    pub version: u8,
    pub rows: Vec<TargetRow>,
    pub prepared: BTreeMap<String, Prepared>,
    pub outcomes: BTreeMap<String, Outcome>,
    pub waiting: BTreeMap<String, WaitingRecord>,
    pub deliveries: BTreeMap<String, super::Delivery>,
}
use super::super::pulls::Kept;

/// Installed before the agent registry opens; native projections read the same
/// target rows whose atomic replacement retains approval, receipt and outbox.
pub struct DraftJournal {
    pub(crate) store: Arc<dyn DurableStore>,
    limits: JournalLimits,
    kept: Mutex<Kept>,
}
impl std::fmt::Debug for DraftJournal {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DraftJournal").finish_non_exhaustive()
    }
}
impl DraftJournal {
    /// Migrates the complete old target log in one durable replacement. Once a
    /// snapshot exists, the old stream is never consulted or written again.
    ///
    /// # Errors
    /// Oversized/corrupt history, failed migration or uncertain storage.
    pub fn open(
        store: Arc<dyn DurableStore>,
        limits: JournalLimits,
    ) -> Result<Arc<Self>, DraftError> {
        let bytes = block_on(store.get_routed(ROUTE, KEY))
            .map_err(|_| DraftError::Store)?
            .map_err(|_| DraftError::Store)?;
        let mut snapshot = Snapshot {
            version: 2,
            ..Snapshot::default()
        };
        if let Some(bytes) = &bytes {
            if bytes.len() > limits.snapshot_bytes.get() {
                return Err(DraftError::Bound);
            }
            snapshot = serde_json::from_slice(bytes).map_err(|_| DraftError::Store)?;
            if snapshot.version != 2 {
                return Err(DraftError::Contract);
            }
        } else {
            Self::migrate(store.as_ref(), limits, &mut snapshot)?;
        }
        let journal = Arc::new(Self {
            store,
            limits,
            kept: Mutex::new(Kept {
                snapshot,
                hash: bytes.as_deref().map(Hash::of),
                uncertain: false,
                active: None,
                pending: None,
            }),
        });
        journal.validate(&journal.lock()?.snapshot)?;
        if bytes.is_none() {
            journal.update(|_| Ok(()))?;
        }
        Ok(journal)
    }
    fn migrate(
        store: &dyn DurableStore,
        limits: JournalLimits,
        snapshot: &mut Snapshot,
    ) -> Result<(), DraftError> {
        let head = block_on(store.read_stream_next_seq(REGISTRY_STREAM))
            .map_err(|_| DraftError::Store)?
            .map_err(|_| DraftError::Store)?
            .unwrap_or(0);
        if head > u64::try_from(limits.rows.get()).map_err(|_| DraftError::Bound)? {
            return Err(DraftError::Bound);
        }
        let mut cursor = Some(ScanCursor {
            stream_key: REGISTRY_STREAM.into(),
            next_sequence: 0,
        });
        let mut next = 0;
        let mut imported_bytes: usize = 0;
        while next < head {
            let page =
                block_on(store.scan_page(REGISTRY_STREAM, cursor, limits.replay_page_rows.get()))
                    .map_err(|_| DraftError::Store)?
                    .map_err(|_| DraftError::Store)?;
            if page.entries.is_empty() {
                return Err(DraftError::Contract);
            }
            cursor = page.next;
            for row in page.entries {
                imported_bytes = imported_bytes
                    .checked_add(row.payload.len())
                    .ok_or(DraftError::Bound)?;
                if imported_bytes > limits.snapshot_bytes.get() {
                    return Err(DraftError::Bound);
                }
                if row.sequence != next || next >= head {
                    return Err(DraftError::Contract);
                }
                let target: Value =
                    serde_json::from_slice(&row.payload).map_err(|_| DraftError::Store)?;
                let approval = Self::legacy_approval(&target)?;
                if let Some(approval) = &approval {
                    let receipt: Execution =
                        serde_json::from_value(target["receipt"]["execution"].clone())
                            .map_err(|_| DraftError::Store)?;
                    if snapshot
                        .outcomes
                        .insert(
                            approval.draft.clone(),
                            Outcome {
                                draft: None,
                                approval: approval.clone(),
                                receipt,
                                acknowledged: false,
                                delivered: false,
                            },
                        )
                        .is_some()
                    {
                        return Err(DraftError::Contract);
                    }
                }
                snapshot.rows.push(TargetRow {
                    timestamp: row.timestamp,
                    target,
                    approval,
                });
                next = next.checked_add(1).ok_or(DraftError::Bound)?;
            }
        }
        if block_on(store.read_stream_next_seq(REGISTRY_STREAM))
            .map_err(|_| DraftError::Store)?
            .map_err(|_| DraftError::Store)?
            .unwrap_or(0)
            != head
        {
            return Err(DraftError::Contract);
        }
        Ok(())
    }
    fn legacy_approval(target: &Value) -> Result<Option<ApprovalTransition>, DraftError> {
        if target["receipt"].is_null() {
            return Ok(None);
        }
        let receipt: Execution = serde_json::from_value(target["receipt"]["execution"].clone())
            .map_err(|_| DraftError::Store)?;
        Ok(Some(ApprovalTransition {
            draft: target["receipt"]["draft"]
                .as_str()
                .ok_or(DraftError::Contract)?
                .into(),
            digest: execution_digest(&receipt).into(),
            approved_by: vec!["lys".into()],
            executed_at_ms: target["transition"]["at_ms"]
                .as_u64()
                .ok_or(DraftError::Contract)?,
        }))
    }
    pub(crate) fn lock(&self) -> Result<MutexGuard<'_, Kept>, DraftError> {
        let kept = self.kept.lock().map_err(|_| DraftError::Store)?;
        if kept.uncertain {
            return Err(DraftError::Store);
        }
        Ok(kept)
    }
    fn validate(&self, snapshot: &Snapshot) -> Result<Vec<u8>, DraftError> {
        if snapshot.rows.len() > self.limits.rows.get()
            || snapshot.prepared.len() > self.limits.operations.get()
            || snapshot.outcomes.len() > self.limits.operations.get()
            || snapshot.deliveries.len() > self.limits.operations.get()
            || snapshot.waiting.len() > self.limits.operations.get()
        {
            return Err(DraftError::Bound);
        }
        let bytes = serde_json::to_vec(snapshot).map_err(|_| DraftError::Store)?;
        if bytes.len() > self.limits.snapshot_bytes.get()
            || self
                .store
                .max_request_bytes()
                .is_some_and(|bound| bytes.len() > bound)
        {
            return Err(DraftError::Bound);
        }
        Ok(bytes)
    }
    /// Keeps a candidate only after one conditional replacement and one flush.
    /// A failed write/flush poisons this owner until it is reopened from disk.
    ///
    /// # Errors
    /// Conflict, capacity or storage failure; no failure is converted to success.
    pub fn update<T>(
        &self,
        change: impl FnOnce(&mut Snapshot) -> Result<T, DraftError>,
    ) -> Result<T, DraftError> {
        self.replace(change, true)
    }
    fn replace<T>(
        &self,
        change: impl FnOnce(&mut Snapshot) -> Result<T, DraftError>,
        flush: bool,
    ) -> Result<T, DraftError> {
        let mut kept = self.lock()?;
        if kept.pending.is_some() || (flush && kept.active.is_some()) {
            return Err(DraftError::Contract);
        }
        let mut candidate = kept.snapshot.clone();
        let result = change(&mut candidate)?;
        let bytes = self.validate(&candidate)?;
        let hash = Hash::of(&bytes);
        let write = block_on(self.store.apply_routed_batch(
            ROUTE,
            vec![RoutedBatchItem {
                key: KEY.into(),
                expected: kept.hash,
                value: bytes,
                ttl: None,
            }],
        ))
        .map_err(|_| DraftError::Store)
        .and_then(|result| result.map_err(|_| DraftError::Store));
        if write.is_err() {
            kept.uncertain = true;
            return Err(DraftError::Store);
        }
        if flush {
            if !matches!(block_on(self.store.flush()), Ok(Ok(()))) {
                kept.uncertain = true;
                return Err(DraftError::Store);
            }
            kept.snapshot = candidate;
            kept.hash = Some(hash);
        } else {
            kept.pending = Some((candidate, hash));
        }
        drop(kept);
        Ok(result)
    }
    /// Reads a bounded durable projection without exposing it in diagnostics.
    ///
    /// # Errors
    /// An uncertain or unavailable journal.
    pub fn read<T>(&self, read: impl FnOnce(&Snapshot) -> T) -> Result<T, DraftError> {
        Ok(read(&self.lock()?.snapshot))
    }
    /// Captures only the matching native agent transition. Other presence
    /// observations remain ordinary target rows, without approval metadata.
    ///
    /// # Errors
    /// Another approval is executing or the journal is unavailable.
    pub fn begin(
        &self,
        draft: &ApprovedDraft,
        approval: &ApprovalTransition,
    ) -> Result<(), DraftError> {
        let mut kept = self.lock()?;
        if approval.draft != draft.id
            || approval.digest != draft.request_digest
            || approval.approved_by.is_empty()
        {
            return Err(DraftError::ReceiptMismatch);
        }
        if kept.active.is_some() {
            return Err(DraftError::Store);
        }
        let act: crate::agents::records::AgentAct =
            serde_json::from_str(&draft.words).map_err(|_| DraftError::Contract)?;
        let mut candidate = kept.snapshot.clone();
        let receipt = Execution::Executed {
            request_digest: draft.request_digest.clone(),
            receipt_digest: lys_pass::drafts::request_digest(&format!(
                "{}:{}:{}",
                draft.id,
                draft.request_digest,
                candidate.rows.len()
            )),
        };
        let target = serde_json::json!({"version":1,"name":act.name,"owner":act.owner,"transition":{"holder":act.holder,"at_ms":approval.executed_at_ms,"change":act.change},"receipt":{"draft":draft.id,"execution":receipt}});
        candidate.rows.push(TargetRow {
            timestamp: approval.executed_at_ms,
            target,
            approval: Some(approval.clone()),
        });
        candidate.outcomes.insert(
            draft.id.clone(),
            Outcome {
                draft: Some(draft.clone()),
                approval: approval.clone(),
                receipt,
                acknowledged: false,
                delivered: false,
            },
        );
        self.validate(&candidate)?;
        kept.active = Some((draft.clone(), approval.clone()));
        drop(kept);
        Ok(())
    }
    /// Removes the synchronous execution context after its target owner returns.
    ///
    /// # Errors
    /// The journal is unavailable.
    pub fn finish(&self) -> Result<(), DraftError> {
        self.kept.lock().map_err(|_| DraftError::Store)?.active = None;
        Ok(())
    }
    pub(crate) fn append_native(
        &self,
        payloads: Vec<Vec<u8>>,
        expected: u64,
    ) -> Result<u64, DraftError> {
        let active = self.lock()?.active.clone();
        self.replace(
            |snapshot| {
                if u64::try_from(snapshot.rows.len()).map_err(|_| DraftError::Bound)? != expected
                    || payloads.is_empty()
                {
                    return Err(DraftError::Contract);
                }
                for payload in payloads {
                    let target: Value =
                        serde_json::from_slice(&payload).map_err(|_| DraftError::Contract)?;
                    let approval = if target["receipt"].is_null() {
                        None
                    } else {
                        let (draft, transition) = active.as_ref().ok_or(DraftError::Contract)?;
                        let receipt: Execution =
                            serde_json::from_value(target["receipt"]["execution"].clone())
                                .map_err(|_| DraftError::Contract)?;
                        if target["receipt"]["draft"] != draft.id
                            || execution_digest(&receipt) != draft.request_digest
                            || target["name"] != draft.target.id
                            || target["transition"]["at_ms"] != transition.executed_at_ms
                            || snapshot.outcomes.contains_key(&draft.id)
                        {
                            return Err(DraftError::ReceiptMismatch);
                        }
                        snapshot.outcomes.insert(
                            draft.id.clone(),
                            Outcome {
                                draft: Some(draft.clone()),
                                approval: transition.clone(),
                                receipt,
                                acknowledged: false,
                                delivered: false,
                            },
                        );
                        Some(transition.clone())
                    };
                    snapshot.rows.push(TargetRow {
                        timestamp: target["transition"]["at_ms"]
                            .as_u64()
                            .ok_or(DraftError::Contract)?,
                        target,
                        approval,
                    });
                }
                u64::try_from(snapshot.rows.len()).map_err(|_| DraftError::Bound)
            },
            false,
        )
    }
    pub(crate) fn flush_pending(&self) -> Result<(), DurabilityError> {
        let mut kept = self.lock().map_err(|error| Self::storage(&error))?;
        let result = block_on(self.store.flush())
            .map_err(|_| Self::storage(&DraftError::Store))
            .and_then(|result| result);
        if let Err(error) = result {
            kept.uncertain = true;
            return Err(error);
        }
        if let Some((snapshot, hash)) = kept.pending.take() {
            kept.snapshot = snapshot;
            kept.hash = Some(hash);
        }
        drop(kept);
        Ok(())
    }
    pub(crate) fn storage(error: &DraftError) -> DurabilityError {
        DurabilityError::AppendUnresolved {
            stream_key: REGISTRY_STREAM.into(),
            sequence: 0,
            cause: error.to_string(),
        }
    }
}

#[path = "waiting_agents.rs"]
mod agents;
pub use agents::{AgentBindings, ApprovalAgents};