liminal-server 0.24.9

//! One atomic event batch and one flush per transition; no history is pruned.

use std::sync::{Arc, Mutex, MutexGuard};

use liminal::durability::{DurableStore, bridge::block_on};
use lys_pass::drafts::{ApprovedDraft, Execution, request_digest};

use super::config::AgentsConfig;
use super::error::AgentError;
use super::records::{AgentAct, AgentEvent, AgentRecord, AgentTransition};

#[path = "registry_storage.rs"]
mod storage;
use crate::services::lifecycle::Placer;
use storage::{
    Index, Land, Receipt, action, apply, check_bytes, check_population, refusal_name, store_error,
    validate_act,
};

/// A new namespace leaves an old installation's service records unchanged.
pub const REGISTRY_STREAM: &str = "liminal/agents/v1";

/// Lys owns every process; this seam cannot execute a program locally.
pub trait AgentSeats: Send + Sync {
    /// Starts a reviewed seat once under `operation`.
    ///
    /// # Errors
    /// Lys's named refusal, without a response body.
    fn start(&self, name: &str, operation: &str) -> Result<(), AgentError>;
    /// Stops that seat once under `operation`.
    ///
    /// # Errors
    /// Lys's named refusal, without a response body.
    fn stop(&self, name: &str, operation: &str) -> Result<(), AgentError>;
}

/// Registry over the server's own durable store. No pass or message is retained.
pub struct AgentRegistry {
    store: Arc<dyn DurableStore>,
    bounds: AgentsConfig,
    network: String,
    placer: Option<Arc<dyn Placer>>,
    seats: Option<Arc<dyn AgentSeats>>,
    index: Mutex<Index>,
}

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

impl AgentRegistry {
    /// Restores every transition in bounded pages, including retired agents.
    /// `placer` and `seats` are production Lys bindings, never authority to
    /// approve a draft. Only the authenticated approved-draft connector may
    /// call [`Self::execute_approved`].
    ///
    /// # Errors
    /// Invalid bounds, corrupt history, or a failed store read.
    pub fn open(
        store: Arc<dyn DurableStore>,
        bounds: AgentsConfig,
        network: String,
        placer: Option<Arc<dyn Placer>>,
        seats: Option<Arc<dyn AgentSeats>>,
    ) -> Result<Arc<Self>, AgentError> {
        bounds.validate()?;
        bounds.field(&network, "network")?;
        let mut index = Index::default();
        loop {
            let page =
                block_on(store.read_from(REGISTRY_STREAM, index.next, bounds.replay_page_rows))
                    .map_err(store_error)?
                    .map_err(store_error)?;
            if page.is_empty() {
                break;
            }
            for row in page {
                if row.sequence != index.next {
                    return Err(AgentError::Invalid("history sequence"));
                }
                check_bytes(&bounds, row.payload.len())?;
                let land: Land = serde_json::from_slice(&row.payload)
                    .map_err(|_| AgentError::Invalid("stored agent event"))?;
                if land.version != 1 {
                    return Err(AgentError::Invalid("stored event version"));
                }
                validate_act(&bounds, &land.name, &land.transition)?;
                bounds.field(&land.owner, "owner")?;
                apply(&mut index, land)?;
            }
        }
        Ok(Arc::new(Self {
            store,
            bounds,
            network,
            placer,
            seats,
            index: Mutex::new(index),
        }))
    }

    fn lock(&self) -> Result<MutexGuard<'_, Index>, AgentError> {
        let index = self
            .index
            .lock()
            .map_err(|_| AgentError::Store("lock poisoned".into()))?;
        if index.uncertain {
            return Err(AgentError::Uncertain);
        }
        Ok(index)
    }

    /// Reads one whole history, without inventing a transition after restart.
    ///
    /// # Errors
    /// The name is unknown or the store outcome is unresolved.
    pub fn record(&self, name: &str) -> Result<AgentRecord, AgentError> {
        self.lock()?
            .records
            .get(name)
            .cloned()
            .ok_or_else(|| AgentError::Unknown(name.into()))
    }

    /// Every record, including retired stations, in name order.
    ///
    /// # Errors
    /// The index is unavailable.
    pub fn records(&self) -> Result<Vec<AgentRecord>, AgentError> {
        Ok(self.lock()?.records.values().cloned().collect())
    }

    /// Compact station metadata; topology never clones a transition history.
    ///
    /// # Errors
    /// The index is unavailable.
    pub fn station_records(&self) -> Result<Vec<AgentRecord>, AgentError> {
        Ok(self
            .lock()?
            .records
            .values()
            .map(|record| AgentRecord {
                name: record.name.clone(),
                harness: record.harness,
                purpose: record.purpose.clone(),
                owner: record.owner.clone(),
                service: record.service.clone(),
                created_at_ms: record.created_at_ms,
                transitions: record.transitions.last().cloned().into_iter().collect(),
            })
            .collect())
    }

    /// Reads the receipt committed in the same batch as its mutation.
    ///
    /// # Errors
    /// The index is unavailable.
    pub fn receipt(&self, draft: &str) -> Result<Option<Execution>, AgentError> {
        Ok(self.lock()?.receipts.get(draft).cloned())
    }

    /// A held response cannot mutate the registry.
    #[must_use]
    pub fn held(&self, action: &str, draft: &str) -> AgentError {
        AgentError::Held {
            action: action.into(),
            draft: draft.into(),
        }
    }

    /// Executes only a draft pulled from Lys by the authenticated connector.
    /// Its target and immutable digest are checked again here. Receipt and
    /// transition share one event, so a crash cannot keep one without the other.
    /// Lys starts and stops use the draft id as their idempotency operation.
    ///
    /// # Errors
    /// Foreign or changed drafts, a bound, Lys's refusal, or an unresolved write.
    pub fn execute_approved(&self, draft: &ApprovedDraft) -> Result<Execution, AgentError> {
        let at_ms = u64::try_from(
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_err(|_| AgentError::Invalid("agent execution clock"))?
                .as_millis(),
        )
        .map_err(|_| AgentError::Invalid("agent execution clock overflow"))?;
        self.execute_approved_at(draft, at_ms)
    }

    /// Uses the execution owner's observed time, not the draft's preparation time.
    ///
    /// # Errors
    /// The same refusals as [`Self::execute_approved`].
    pub fn execute_approved_at(
        &self,
        draft: &ApprovedDraft,
        at_ms: u64,
    ) -> Result<Execution, AgentError> {
        self.bounds.field(&draft.id, "draft")?;
        self.bounds.field(&draft.grant, "grant")?;
        check_bytes(&self.bounds, draft.words.len())?;
        if draft.app != "liminal"
            || draft.target.kind != "liminal.agent"
            || request_digest(&draft.words) != draft.request_digest
        {
            return Err(AgentError::Invalid("approved draft target or digest"));
        }
        let act: AgentAct = serde_json::from_str(&draft.words)
            .map_err(|_| AgentError::Invalid("approved agent act"))?;
        if act.name != draft.target.id || action(&act.change) != Some(draft.target.action.as_str())
        {
            return Err(AgentError::Invalid("approved agent action"));
        }
        let transition = AgentTransition {
            holder: act.holder,
            at_ms,
            change: act.change,
        };
        validate_act(&self.bounds, &act.name, &transition)?;
        self.bounds.field(&act.owner, "owner")?;
        let mut index = self.lock()?;
        if let Some(receipt) = index.receipts.get(&draft.id) {
            return match receipt {
                Execution::Executed { request_digest, .. }
                | Execution::RefusedOnExecution { request_digest, .. }
                    if request_digest == &draft.request_digest =>
                {
                    Ok(receipt.clone())
                }
                _ => Err(AgentError::DraftMismatch),
            };
        }
        Self::prepare(&index, &act.name, &transition.change)?;
        if !matches!(transition.change, AgentEvent::Create(_))
            && index
                .records
                .get(&act.name)
                .is_some_and(|record| record.owner != act.owner)
        {
            return Err(AgentError::Invalid(
                "agent owner differs from its prepared act",
            ));
        }
        if matches!(transition.change, AgentEvent::Create(_)) {
            check_population(&index, &self.bounds, &act.owner)?;
        }
        let receipt = Execution::Executed {
            request_digest: draft.request_digest.clone(),
            receipt_digest: request_digest(&format!(
                "{}:{}:{}",
                draft.id, draft.request_digest, index.next
            )),
        };
        let land = Land {
            version: 1,
            name: act.name.clone(),
            owner: act.owner,
            transition: transition.clone(),
            receipt: Some(Receipt {
                draft: draft.id.clone(),
                execution: receipt.clone(),
            }),
        };
        let prepared =
            serde_json::to_vec(&land).map_err(|_| AgentError::Invalid("event encoding"))?;
        check_bytes(&self.bounds, prepared.len())?;
        match &transition.change {
            AgentEvent::Create(_) => self.place(&act.name, &draft.id)?,
            AgentEvent::Start => self
                .seats
                .as_ref()
                .ok_or(AgentError::Invalid("Lys seat binding is not configured"))?
                .start(&act.name, &draft.id)?,
            AgentEvent::Stop => self
                .seats
                .as_ref()
                .ok_or(AgentError::Invalid("Lys seat binding is not configured"))?
                .stop(&act.name, &draft.id)?,
            AgentEvent::Retire => {
                if index
                    .records
                    .get(&act.name)
                    .is_some_and(|record| record.connection().is_some())
                {
                    return Err(AgentError::Transition("stop and close before retirement"));
                }
            }
            _ => return Err(AgentError::Invalid("presence is observed by the server")),
        }
        self.land(&mut index, land)?;
        Ok(receipt)
    }

    /// Records a connection observation after identity and rights were decided.
    ///
    /// # Errors
    /// Not a presence event, a stale connection, or a store failure.
    pub(crate) fn observe(
        &self,
        name: &str,
        transition: AgentTransition,
    ) -> Result<(), AgentError> {
        if !matches!(
            transition.change,
            AgentEvent::Present { .. }
                | AgentEvent::Gone { .. }
                | AgentEvent::Session { .. }
                | AgentEvent::SessionClosed { .. }
        ) {
            return Err(AgentError::Invalid("server observation"));
        }
        validate_act(&self.bounds, name, &transition)?;
        let mut index = self.lock()?;
        Self::prepare(&index, name, &transition.change)?;
        let owner = index
            .records
            .get(name)
            .ok_or_else(|| AgentError::Unknown(name.into()))?
            .owner
            .clone();
        let result = self.land(
            &mut index,
            Land {
                version: 1,
                name: name.into(),
                owner,
                transition,
                receipt: None,
            },
        );
        drop(index);
        result
    }

    fn prepare(index: &Index, name: &str, change: &AgentEvent) -> Result<(), AgentError> {
        if let AgentEvent::Create(_) = change {
            if index.records.contains_key(name) {
                return Err(AgentError::NameTaken(name.into()));
            }
        } else {
            let record = index
                .records
                .get(name)
                .ok_or_else(|| AgentError::Unknown(name.into()))?;
            if !record.active() {
                return Err(AgentError::Transition("agent is retired"));
            }
            if let AgentEvent::Present { connection } = change {
                if let Some((existing, since_ms)) = record.seat() {
                    return Err(AgentError::Present {
                        agent: name.into(),
                        connection: existing,
                        since_ms,
                    });
                }
                if *connection == 0 {
                    return Err(AgentError::Invalid("connection"));
                }
            }
            if let AgentEvent::Gone { connection, .. } = change {
                if record.connection() != Some(*connection) {
                    return Err(AgentError::Transition("departure names another connection"));
                }
            }
        }
        Ok(())
    }

    fn place(&self, name: &str, draft: &str) -> Result<(), AgentError> {
        let placer = self
            .placer
            .as_ref()
            .ok_or(AgentError::Invalid("agent placer is not configured"))?;
        let subjects = [
            format!("agents.{name}.inbox"),
            format!("agents.{name}.presence"),
            format!("agents.{name}.state"),
        ];
        for child in std::iter::once(("liminal.agent", name)).chain(
            subjects
                .iter()
                .map(|subject| ("liminal.subject", subject.as_str())),
        ) {
            // Lys's operation id is `op-` and exactly 32 hex digits.
            let digest =
                blake3::hash(format!("agent-place:{draft}:{}:{}", child.0, child.1).as_bytes());
            let operation = format!("op-{}", &digest.to_hex()[..32]);
            placer
                .place(&operation, child, ("liminal.network", &self.network))
                .map_err(|refusal| AgentError::Store(refusal_name(&refusal)))?;
        }
        Ok(())
    }

    fn land(&self, index: &mut Index, land: Land) -> Result<(), AgentError> {
        if let AgentEvent::Create(_) = &land.transition.change {
            check_population(index, &self.bounds, &land.owner)?;
        }
        let bytes = serde_json::to_vec(&land).map_err(|_| AgentError::Invalid("event encoding"))?;
        check_bytes(&self.bounds, bytes.len())?;
        let appended = block_on(
            self.store
                .append_all(REGISTRY_STREAM, vec![bytes], index.next),
        )
        .map_err(store_error)
        .and_then(|answer| answer.map_err(store_error));
        match appended {
            Ok(next) if Some(next) == index.next.checked_add(1) => {}
            other => {
                tracing::warn!(expected = index.next, answer = ?other.map_err(|error| error.to_string()), "agent registry append outcome unknown");
                index.uncertain = true;
                return Err(AgentError::Uncertain);
            }
        }
        if block_on(self.store.flush())
            .map_err(store_error)
            .and_then(|answer| answer.map_err(store_error))
            .is_err()
        {
            index.uncertain = true;
            return Err(AgentError::Uncertain);
        }
        apply(index, land)
    }
}