liminal-server 0.24.1

//! The server's registry and retained subjects, sharing its store and scheduler.

use std::sync::{Arc, OnceLock};

use liminal::channel::{ChannelSupervisor, InboxInstall};
use liminal::durability::{DurableStore, bridge::block_on};
use liminal::protocol::SchemaId;
use serde_json::json;

use super::{
    binding,
    config::AgentsConfig,
    error::AgentError,
    presence::Presence,
    records::{AgentEvent, AgentTransition},
    registry::{AgentRegistry, AgentSeats, REGISTRY_STREAM},
    retained::Retained,
};
use crate::auth_lys::{HotAction, session::LysSession};
use crate::config::types::ServerConfig;
use crate::console::wall::WallBase;
use crate::server::connection::services::{ConnectionSubscription, LiminalConnectionServices};
use crate::services::lifecycle::Placer;

/// Agent metadata never hosts a provider, harness pipe or process.
#[derive(Debug)]
pub struct Agents {
    registry: Arc<AgentRegistry>,
    retained: Arc<Retained>,
    presence: Presence,
    bounds: AgentsConfig,
    network: String,
    wall: WallBase,
    listens: OnceLock<Arc<super::listens::Listens>>,
}

/// One subscription request with the connection's verified identity.
pub struct AgentSubscription<'a> {
    /// The server's ordinary channel adapter.
    pub services: &'a LiminalConnectionServices,
    /// The connection process id.
    pub connection: u64,
    /// The admitted pass session.
    pub session: Option<&'a dyn LysSession>,
    /// The exact listened subject.
    pub subject: &'a str,
    /// Schemas the client accepts.
    pub schemas: &'a [SchemaId],
    /// The connection's installed inbox bounds and notifier.
    pub install: Option<InboxInstall>,
    /// A server-allocated subscription id.
    pub id: u64,
}

impl Agents {
    /// Binds the console once; subscription resources never keep it alive.
    ///
    /// # Errors
    /// A console was already installed.
    pub fn bind_console(
        &self,
        console: &Arc<crate::console::ConsoleHub>,
    ) -> Result<(), AgentError> {
        self.listens
            .set(super::listens::Listens::new(self.bounds, console))
            .map_err(|_| AgentError::Invalid("agent console already bound"))?;
        self.retained.bind_console(console)?;
        Ok(())
    }

    /// Exact subjects of the live subscriptions, each once.
    ///
    /// # Errors
    /// The listen index is unavailable.
    pub fn listened_subjects(&self, connection: u64) -> Result<Vec<String>, AgentError> {
        self.listens
            .get()
            .map_or(Ok(Vec::new()), |listens| listens.subjects(connection))
    }

    /// Wraps the real resource so unsubscribe, overflow and Drop remove its stop.
    ///
    /// # Errors
    /// A listen bound or index failure.
    pub fn track_listen(
        &self,
        connection: u64,
        id: u64,
        subject: &str,
        resource: Box<dyn crate::server::connection::services::SubscriptionResource>,
    ) -> Result<Box<dyn crate::server::connection::services::SubscriptionResource>, AgentError>
    {
        if self.presence.agent(connection)?.is_none() {
            return Ok(resource);
        }
        self.listens
            .get()
            .ok_or(AgentError::Invalid("agent console is not bound"))?
            .attach(connection, id, subject, resource)
    }
    /// Applies an agent draft pulled and authenticated by AGT-004, then seeds
    /// the retained subjects. Replays reuse the committed mutation receipt.
    ///
    /// # Errors
    /// Draft, bound, store or retained-subject failure.
    pub fn execute_approved(
        &self,
        draft: &lys_pass::drafts::ApprovedDraft,
    ) -> Result<lys_pass::drafts::Execution, AgentError> {
        let receipt = self.registry.execute_approved_at(draft, self.now()?)?;
        if draft.target.action == "create_agent" {
            self.created(&draft.target.id)?;
        }
        Ok(receipt)
    }
    /// Opens over the server's own store and channel scheduler. Previously
    /// present tickets are durably recorded gone before admission opens.
    ///
    /// # Errors
    /// Invalid configuration, corrupt history, or a store or clock failure.
    pub fn open(
        store: Arc<dyn DurableStore>,
        bounds: AgentsConfig,
        network: String,
        supervisor: ChannelSupervisor,
        device: &str,
        placer: Option<Arc<dyn Placer>>,
        seats: Option<Arc<dyn AgentSeats>>,
    ) -> Result<Arc<Self>, AgentError> {
        let registry = AgentRegistry::open(store, bounds, network.clone(), placer, seats)?;
        let retained = Retained::new(bounds, supervisor, device);
        let wall = WallBase::now().map_err(|_| AgentError::Invalid("agent wall clock"))?;
        let now = wall
            .unix_ms_at(std::time::Instant::now())
            .map_err(|_| AgentError::Invalid("agent wall clock"))?;
        for record in registry.records()? {
            if let Some(connection) = record.connection() {
                registry.observe(
                    &record.name,
                    AgentTransition {
                        holder: "server".into(),
                        at_ms: now,
                        change: AgentEvent::Gone {
                            connection,
                            reason: "server_restart".into(),
                        },
                    },
                )?;
            }
            let current = registry.record(&record.name)?;
            let last_gone =
                current
                    .transitions
                    .iter()
                    .rev()
                    .find_map(|transition| match &transition.change {
                        AgentEvent::Gone { reason, .. } => {
                            Some((transition.at_ms, reason.as_str()))
                        }
                        _ => None,
                    });
            let (at_ms, reason) = last_gone.unwrap_or((current.created_at_ms, "created"));
            retained.publish(&[
                (
                    format!("agents.{}.presence", record.name),
                    json!({"present":false,"at_ms":at_ms,"reason":reason}),
                ),
                (
                    format!("agents.{}.state", record.name),
                    json!({"state":"gone"}),
                ),
            ])?;
        }
        Ok(Arc::new(Self {
            registry,
            retained,
            presence: Presence::default(),
            bounds,
            network,
            wall,
            listens: OnceLock::new(),
        }))
    }

    /// The one registry shared with the approved-draft executor.
    #[must_use]
    pub const fn registry(&self) -> &Arc<AgentRegistry> {
        &self.registry
    }
    /// The one retained-subject writer shared with approvals.
    #[must_use]
    pub const fn retained(&self) -> &Arc<Retained> {
        &self.retained
    }
    /// The configured positive bounds.
    #[must_use]
    pub const fn bounds(&self) -> &AgentsConfig {
        &self.bounds
    }
    /// The network whose placed resources decide these subjects.
    #[must_use]
    pub fn network(&self) -> &str {
        &self.network
    }
    /// The observed presence, never inferred from a lifecycle request.
    #[must_use]
    pub const fn presence(&self) -> &Presence {
        &self.presence
    }

    /// Seeds subjects after a draft created a station. Existing presence is
    /// preserved; a second executor notification cannot mark a live seat gone.
    ///
    /// # Errors
    /// Unknown station or failed retained delivery.
    pub fn created(&self, name: &str) -> Result<(), AgentError> {
        let record = self.registry.record(name)?;
        if self.presence.connection(name)?.is_none() {
            self.retained.publish(&[
                (
                    format!("agents.{name}.presence"),
                    json!({"present":false,"at_ms":record.created_at_ms,"reason":"created"}),
                ),
                (format!("agents.{name}.state"), json!({"state":"gone"})),
            ])?;
        }
        Ok(())
    }

    /// Opens the real inbox or a retained metadata watcher after both identity
    ///
    /// and subject rights are decided. The caller's revocation and membership
    /// checks still precede this method and still run at delivery release.
    ///
    /// # Errors
    /// The pass, station, duplicate seat, schema or subscription is refused.
    pub fn subscribe(
        &self,
        request: AgentSubscription<'_>,
        ordinary: impl FnOnce(Option<InboxInstall>) -> Result<ConnectionSubscription, AgentError>,
    ) -> Result<ConnectionSubscription, AgentError> {
        let AgentSubscription {
            services,
            connection,
            session,
            subject,
            schemas,
            install,
            id,
        } = request;
        let now = self.now()?;
        binding::decide(session, &self.network, HotAction::Subscribe, subject, now)?;
        let (name, leaf) = binding::subject(subject).ok_or(AgentError::Invalid("agent subject"))?;
        if matches!(leaf, "presence" | "state") {
            self.registry.record(name)?;
            return self.retained.subscribe(id, subject, schemas, install);
        }
        if !self.registry.record(name)?.active() {
            return Err(AgentError::Transition("agent is retired"));
        }
        if leaf != "inbox" {
            return ordinary(install);
        }
        let session = session.ok_or(AgentError::Invalid("agent pass"))?;
        let subscription = self.presence.connect(
            &self.registry,
            &self.retained,
            super::presence::ConnectionObservation {
                name,
                holder: session.holder(),
                connection,
                at_ms: now,
            },
            || ordinary(install),
        )?;
        services
            .console()
            .emit(crate::console::TopologyChange::Service {
                subject: subject.into(),
                present: true,
            })
            .map_err(|_| AgentError::Delivery("agent topology change failed"))?;
        Ok(subscription)
    }

    /// A transport's named close reaches the same durable observation path.
    ///
    /// # Errors
    /// A store, clock or delivery failure; no automatic retry is added.
    pub fn close(&self, connection: u64, reason: &str) -> Result<(), AgentError> {
        self.bounds.field(reason, "transport reason")?;
        if let Some(listens) = self.listens.get() {
            listens.close(connection)?;
        }
        self.presence.close(
            &self.registry,
            &self.retained,
            connection,
            reason,
            self.now()?,
        )
    }

    pub(crate) fn now(&self) -> Result<u64, AgentError> {
        self.wall
            .unix_ms_at(std::time::Instant::now())
            .map_err(|_| AgentError::Invalid("agent wall clock"))
    }
}

/// Checks the durable namespace even when the configuration block is absent.
/// An old installation with no agent rows keeps its existing behaviour.
///
/// # Errors
/// A placed agent without `[auth.lys.agents]`, or a store read failure.
pub fn configured(
    config: &ServerConfig,
    store: Arc<dyn DurableStore>,
    supervisor: ChannelSupervisor,
    device: &str,
    runtime: Option<tokio::runtime::Handle>,
) -> Result<Option<Arc<Agents>>, crate::ServerError> {
    let lys = config.auth.as_ref().and_then(|auth| auth.lys.as_ref());
    let bounds = lys.and_then(|lys| lys.agents);
    let Some(bounds) = bounds else {
        let head = block_on(store.read_stream_next_seq(REGISTRY_STREAM))
            .map_err(|error| AgentError::Store(error.to_string()))?
            .map_err(|error| AgentError::Store(error.to_string()))?;
        if head.is_some() {
            return Err(crate::ServerError::ConfigValidation {
                message: "[auth.lys.agents] is required when a liminal.agent is placed".into(),
            });
        }
        return Ok(None);
    };
    let lys = lys.ok_or(AgentError::Invalid("Lys configuration"))?;
    let (placer, seats) = super::effects::configured(lys, runtime)?;
    Ok(Some(Agents::open(
        store,
        bounds,
        lys.network_id.clone(),
        supervisor,
        device,
        placer,
        seats,
    )?))
}

/// Records a connection-fate close by the cause its class names: a client's
/// `Disconnect` is `client_close`, as on the ordinary close path.
pub fn connection_fate_closed(
    services: &dyn crate::server::connection::services::ConnectionServices,
    connection: u64,
    class: crate::server::participant::ConnectionFateClass,
) {
    use crate::server::participant::ConnectionFateClass;
    let reason = match class {
        ConnectionFateClass::CleanDisconnect => "client_close",
        ConnectionFateClass::ServerShutdown => "server_shutdown",
        ConnectionFateClass::ConnectionLost => "transport_drop",
        ConnectionFateClass::ProtocolError => "protocol_error",
    };
    connection_closed(services, connection, reason);
}

/// Records the transport's final stable cause. The caller supplies the cause
///
/// it actually observed; the Drop backstop uses `transport_drop`. No payload
/// or credential is retained in a diagnostic if the durable write fails.
pub fn connection_closed(
    services: &dyn crate::server::connection::services::ConnectionServices,
    connection: u64,
    reason: &str,
) {
    if let Some(agents) = services.agents() {
        if let Err(error) = agents.close(connection, reason) {
            tracing::error!(connection, %error, "agent_presence_departure_failed");
        }
    }
}