liminal_server/agents/

server.rs

//! The connection adapter's agent hooks; kept beside agents to limit shared edits.

use super::{
    ChannelAccessError, ChannelRegistration, ConnectionServices, ConnectionSubscription,
    LiminalConnectionServices, ProtocolSchemaId, ServerError,
};
use crate::agents::{binding, error::AgentError};
use crate::auth_lys::session::LysSession;

impl LiminalConnectionServices {
    pub(super) fn configured_agents(
        config: &crate::config::ServerConfig,
        store: std::sync::Arc<dyn liminal::durability::DurableStore>,
        supervisor: liminal::channel::ChannelSupervisor,
        console: &std::sync::Arc<crate::console::ConsoleHub>,
        runtime: Option<tokio::runtime::Handle>,
    ) -> Result<AgentOwners, ServerError> {
        let lys = config.auth.as_ref().and_then(|auth| auth.lys.as_ref());
        let (agents, owners, delivery) = if let Some((lys, bounds, limits)) =
            lys.and_then(|lys| Some((lys, lys.agents?, lys.drafts?)))
        {
            let (placer, seats) = crate::agents::effects::configured(lys, runtime)?;
            let bindings = crate::agents::waiting::durable::AgentBindings {
                bounds,
                network: lys.network_id.clone(),
                supervisor,
                device: console.device().into(),
                placer,
                seats,
            };
            let owners = crate::agents::waiting::durable::ApprovalAgents::open(
                store,
                limits.journal,
                bindings,
            )
            .map_err(draft_error)?;
            let delivery = crate::agents::waiting::AgentDelivery::new(
                &owners,
                std::collections::BTreeMap::new(),
                format!("liminal.{}", console.device()),
                limits.waiting,
            )
            .map_err(draft_error)?;
            (Some(owners.agents().clone()), Some(owners), Some(delivery))
        } else {
            (
                crate::agents::runtime::configured(
                    config,
                    store,
                    supervisor,
                    console.device(),
                    runtime,
                )?,
                None,
                None,
            )
        };
        if let Some(agents) = &agents {
            agents.bind_console(console)?;
        }
        Ok((agents, owners, delivery))
    }
    pub(super) fn publish_agent(
        &self,
        connection: u64,
        session: Option<&dyn LysSession>,
        channel: &str,
        envelope: &liminal::protocol::MessageEnvelope,
        key: Option<&str>,
    ) -> Result<super::PublishOutcome, ServerError> {
        binding::check(session, crate::auth_lys::HotAction::Publish, channel)?;
        if let Some(session) = session.filter(|session| session.holder_kind() == Some("agent")) {
            self.console
                .tap()
                .name_station(connection, session.holder())
                .map_err(|_| AgentError::Delivery("publisher identity projection failed"))?;
        }
        let Some((name, leaf)) = binding::subject(channel) else {
            return ConnectionServices::publish(self, connection, channel, envelope, key);
        };
        let agents = self
            .agents
            .as_ref()
            .ok_or(AgentError::Invalid(crate::agents::error::NO_REGISTRY))?;
        binding::decide(
            session,
            agents.network(),
            crate::auth_lys::HotAction::Publish,
            channel,
            agents.now()?,
        )?;
        if !agents.registry().record(name)?.active() {
            return Err(AgentError::Transition("agent is retired").into());
        }
        if leaf == "inbox" {
            validate_inbox(agents.bounds(), &envelope.payload)?;
            check_sender(&envelope.payload, session.map(LysSession::holder))?;
            self.ensure_agent_channel(channel)?;
        }
        self.publish_with_origin(
            channel,
            envelope,
            key,
            crate::server::self_streams::events::PublishOrigin::Connection { id: connection },
        )
        .map(|publication| publication.outcome)
    }

    /// Executes a pulled agent draft and applies its runtime projection. The
    /// durable receipt is kept even when projection fails, so an explicit
    /// replay repairs projection without starting or creating anything twice.
    ///
    /// # Errors
    /// Registry, Lys effect, roster or retained-subject failure.
    pub fn execute_agent_draft(
        &self,
        draft: &lys_pass::drafts::ApprovedDraft,
    ) -> Result<lys_pass::drafts::Execution, ServerError> {
        let agents = self
            .agents
            .as_ref()
            .ok_or(AgentError::Invalid(crate::agents::error::NO_REGISTRY))?;
        let receipt = agents.execute_approved(draft)?;
        match draft.target.action.as_str() {
            "create_agent" => self.agent_created(&draft.target.id)?,
            "retire" => self.retire_agent_inbox(&draft.target.id)?,
            _ => {}
        }
        Ok(receipt)
    }

    fn retire_agent_inbox(&self, name: &str) -> Result<(), ServerError> {
        let inbox = format!("agents.{name}.inbox");
        let configured = self
            .read_roster()
            .map_err(|error| AgentError::Store(error.to_string()))?
            .get(&inbox)
            .cloned();
        if let Some(configured) = configured {
            self.quiesce_channel(&inbox, "agent_retired")
                .map_err(|error| AgentError::Store(error.to_string()))?;
            configured
                .handle
                .close()
                .map_err(|_| AgentError::Delivery("retired inbox close failed"))?;
            self.write_roster()
                .map_err(|error| AgentError::Store(error.to_string()))?
                .remove(&inbox);
        }
        self.console
            .emit(crate::console::TopologyChange::Service {
                subject: inbox,
                present: false,
            })
            .map_err(|_| AgentError::Delivery("agent retirement topology failed"))?;
        Ok(())
    }
    /// Installs synthetic approved fixtures on the real store and scheduler.
    ///
    /// # Errors
    /// Returns registry, clock or retained projection failures.
    #[cfg(any(test, feature = "test-support"))]
    pub fn with_agent_fixture(
        mut self,
        network: &str,
        placer: std::sync::Arc<dyn crate::services::lifecycle::Placer>,
    ) -> Result<Self, ServerError> {
        self.agents = Some(crate::agents::runtime::Agents::open(
            self.durable_store(),
            crate::agents::config::AgentsConfig::default(),
            network.into(),
            self.cluster.supervisor().clone(),
            self.console.device(),
            Some(placer),
            None,
        )?);
        self.agents
            .as_ref()
            .ok_or(AgentError::Invalid("agent fixture missing"))?
            .bind_console(&self.console)?;
        Ok(self)
    }
    /// Gives every console tap attached from now on a ring of `capacity`,
    /// so a fixture can fill one with a handful of publishes. Call it before
    /// any tap attaches.
    #[cfg(any(test, feature = "test-support"))]
    #[must_use]
    pub fn with_console_tap_capacity(self, capacity: usize) -> Self {
        self.console.set_tap_capacity(capacity);
        self
    }
    pub(super) fn subscribe_agent(
        &self,
        connection: u64,
        session: Option<&dyn LysSession>,
        channel: &str,
        schemas: &[ProtocolSchemaId],
        install: Option<liminal::channel::InboxInstall>,
    ) -> Result<ConnectionSubscription, ServerError> {
        binding::check(session, crate::auth_lys::HotAction::Subscribe, channel)?;
        if binding::subject(channel).is_none() {
            let subscription = ConnectionServices::subscribe(self, channel, schemas, install)?;
            return self.track_agent_listen(connection, channel, subscription);
        }
        let agents = self
            .agents
            .as_ref()
            .ok_or(AgentError::Invalid(crate::agents::error::NO_REGISTRY))?;
        self.place_own_station(session, channel)?;
        self.ensure_agent_channel(channel)?;
        let id = self
            .next_subscription_id
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let subscription = agents
            .subscribe(
                crate::agents::runtime::AgentSubscription {
                    services: self,
                    connection,
                    session,
                    subject: channel,
                    schemas,
                    install,
                    id,
                },
                |install| {
                    self.open_subscription(channel, schemas, install, None)
                        .map_err(|error| AgentError::Store(error.to_string()))
                },
            )
            .map_err(ServerError::from)?;
        self.track_agent_listen(connection, channel, subscription)
    }

    fn track_agent_listen(
        &self,
        connection: u64,
        subject: &str,
        subscription: ConnectionSubscription,
    ) -> Result<ConnectionSubscription, ServerError> {
        let Some(agents) = &self.agents else {
            return Ok(subscription);
        };
        let ConnectionSubscription {
            id,
            channel,
            stream_id,
            selected_schema,
            resource,
            issued,
        } = subscription;
        let resource = agents.track_listen(connection, id, subject, resource)?;
        Ok(ConnectionSubscription {
            id,
            channel,
            stream_id,
            selected_schema,
            resource,
            issued,
        })
    }

    /// Makes an approved station's ordinary inbox available and seeds metadata.
    /// The draft executor calls this after its durable receipt; it grants nothing.
    ///
    /// # Errors
    /// Missing configuration, unknown agent, roster bound, or metadata failure.
    pub fn agent_created(&self, name: &str) -> Result<(), ServerError> {
        self.ensure_agent_channel(&format!("agents.{name}.inbox"))?;
        self.ensure_agent_channel(&format!("agents.{name}.cancel"))?;
        self.ensure_agent_channel(&format!("agents.{name}.stream"))?;
        if let Some(delivery) = &self.agent_delivery {
            let inbox = self
                .read_roster()
                .map_err(|error| AgentError::Store(error.to_string()))?
                .get(&format!("agents.{name}.inbox"))
                .map(|channel| channel.handle.clone())
                .ok_or(AgentError::Invalid("registered inbox absent"))?;
            delivery.bind_inbox(name, inbox).map_err(draft_error)?;
        }
        self.agents
            .as_ref()
            .ok_or(AgentError::Invalid(crate::agents::error::NO_REGISTRY))?
            .created(name)?;
        self.console
            .emit(crate::console::TopologyChange::Service {
                subject: format!("agents.{name}.inbox"),
                present: true,
            })
            .map_err(|_| AgentError::Delivery("agent topology change failed"))?;
        Ok(())
    }

    /// ACCESS-013 R1: a verified agent listening its own inbox with no
    /// station places it, then makes its channels as an approved create does.
    /// The grant decision already ran (`binding::check`), so this never widens
    /// what the pass may do.
    pub(super) fn place_own_station(
        &self,
        session: Option<&dyn LysSession>,
        channel: &str,
    ) -> Result<(), ServerError> {
        let (Some(session), Some((name, "inbox")), Some(agents)) =
            (session, binding::subject(channel), self.agents.as_ref())
        else {
            return Ok(());
        };
        let owner = match session.holder_kind() {
            Some("agent") if session.holder() == name => name.to_owned(),
            Some(crate::agents::open_session::OPEN_KIND) if session.holder() == name => {
                if name == crate::server::connection::declared_holder::ANONYMOUS {
                    return Err(AgentError::Invalid(crate::agents::error::ANONYMOUS_SEAT).into());
                }
                format!(
                    "{}{name}",
                    crate::server::connection::declared_holder::PREFIX
                )
            }
            _ => return Ok(()),
        };
        if agents.registry().place_own(name, &owner, agents.now()?)? {
            self.agent_created(name)?;
        }
        Ok(())
    }

    fn ensure_agent_channel(&self, channel: &str) -> Result<(), ServerError> {
        let (name, leaf) = binding::subject(channel).ok_or(AgentError::Invalid("agent subject"))?;
        let agents = self
            .agents
            .as_ref()
            .ok_or(AgentError::Invalid(crate::agents::error::NO_REGISTRY))?;
        let record = agents
            .registry()
            .record(name)
            .map_err(|error| match error {
                AgentError::Unknown(name) => AgentError::NoStation(name),
                other => other,
            })?;
        if matches!(leaf, "presence" | "state") {
            return Ok(());
        }
        if !record.active() {
            return Err(AgentError::Transition("agent is retired").into());
        }
        if !matches!(leaf, "inbox" | "cancel" | "stream") {
            return Ok(());
        }
        let schema = serde_json::to_vec(&serde_json::json!({"type":"object"}))
            .map_err(|_| AgentError::Invalid("inbox schema encoding"))?;
        self.register_channel(&ChannelRegistration {
            name: channel.into(),
            schema_bytes: (leaf == "inbox").then_some(schema),
            durable: leaf == "inbox",
        })
        .map_err(|error| AgentError::Store(error.to_string()))?;
        Ok(())
    }

    pub(super) fn agent_channel_admission(
        &self,
        operation: crate::server::connection::ChannelOperation,
        channel: &str,
    ) -> Option<Result<(), ChannelAccessError>> {
        let (name, leaf) = binding::subject(channel)?;
        if !matches!(leaf, "presence" | "state" | "inbox" | "cancel" | "stream") {
            return None;
        }
        // ACCESS-013 R1: a subscribe to an inbox with no station yet is
        // decided where the pass is, in `subscribe_agent`: its own holder
        // places it there, and any other name is refused there by name.
        if operation == crate::server::connection::ChannelOperation::Subscribe
            && leaf == "inbox"
            && self
                .agents
                .as_ref()
                .is_some_and(|agents| agents.registry().record(name).is_err())
        {
            return Some(Ok(()));
        }
        Some(self.ensure_agent_channel(channel).map_err(|error| {
            ChannelAccessError::RosterUnavailable {
                message: error.to_string(),
            }
        }))
    }

    /// Keeps agent station metadata in the topology published by the ordinary hub.
    ///
    /// # Errors
    /// A topology or registry record cannot be encoded.
    pub fn agent_topology_payload<'a>(
        &self,
        subject: &str,
        payload: &'a [u8],
    ) -> Result<std::borrow::Cow<'a, [u8]>, crate::console::ConsoleError> {
        let Some(agents) = &self.agents else {
            return Ok(std::borrow::Cow::Borrowed(payload));
        };
        if subject != crate::console::TOPOLOGY_SUBJECT {
            return Ok(std::borrow::Cow::Borrowed(payload));
        }
        let bytes = crate::console::topology::with_agents(payload, agents, self.console.device())
            .map_err(|error| crate::console::ConsoleError::Server {
            subject: subject.into(),
            source: error.into(),
        })?;
        Ok(std::borrow::Cow::Owned(bytes))
    }
}

fn validate_inbox(
    bounds: &crate::agents::config::AgentsConfig,
    payload: &[u8],
) -> Result<(), AgentError> {
    if payload.len() > bounds.max_event_bytes {
        return Err(AgentError::Bound {
            bound: "max_event_bytes",
            limit: bounds.max_event_bytes,
            count: payload.len(),
        });
    }
    let value: serde_json::Value =
        serde_json::from_slice(payload).map_err(|_| AgentError::Invalid("inbox JSON"))?;
    // A prompt is {text}; {command} is a harness command; {config} and {session} read or change
    // the harness's own options and conversations. Each control holds only text and switches.
    let control = |value: &serde_json::Value, keys: &[&str]| {
        value.as_object().is_some_and(|fields| {
            fields.iter().all(|(key, value)| {
                keys.contains(&key.as_str()) && (value.is_string() || value.is_boolean())
            })
        })
    };
    // The sender may be named beside the one message; the exchange overwrites it (CON-022 R3).
    let accepted = value.as_object().is_some_and(|object| {
        let named = object.get("from").map(serde_json::Value::is_string);
        named != Some(false)
            && object.len() == 1 + usize::from(named.is_some())
            && (object.get("text").is_some_and(serde_json::Value::is_string)
                || object
                    .get("config")
                    .is_some_and(|c| control(c, &["option", "value"]))
                || object
                    .get("command")
                    .is_some_and(|c| control(c, &["name", "input"]))
                || object
                    .get("session")
                    .is_some_and(|c| control(c, &["list", "new", "open"])))
    });
    if !accepted {
        return Err(AgentError::Invalid(
            "inbox accepts exactly one of {text}, {command}, {config} or {session}",
        ));
    }
    Ok(())
}

/// The sender an inbox message names, checked against the session's holder (CON-022 R3).
///
/// # Errors
/// The message is not JSON, or names a sender that is not the holder.
fn check_sender(payload: &[u8], holder: Option<&str>) -> Result<(), AgentError> {
    let value: serde_json::Value =
        serde_json::from_slice(payload).map_err(|_| AgentError::Invalid("inbox JSON"))?;
    match (
        value.get("from").and_then(serde_json::Value::as_str),
        holder,
    ) {
        (None, _) => Ok(()),
        (Some(claimed), Some(holder)) if claimed == holder => Ok(()),
        _ => Err(AgentError::Invalid(
            "inbox from names a sender other than the session's holder",
        )),
    }
}

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

type AgentOwners = (
    Option<std::sync::Arc<crate::agents::runtime::Agents>>,
    Option<std::sync::Arc<crate::agents::waiting::durable::ApprovalAgents>>,
    Option<std::sync::Arc<crate::agents::waiting::AgentDelivery>>,
);
fn draft_error(error: crate::auth_lys::drafts::DraftError) -> ServerError {
    error.into()
}
impl From<crate::auth_lys::drafts::DraftError> for ServerError {
    fn from(error: crate::auth_lys::drafts::DraftError) -> Self {
        Self::ConfigValidation {
            message: error.to_string(),
        }
    }
}