liminal_server/agents/
runtime.rs

1//! The server's registry and retained subjects, sharing its store and scheduler.
2
3use std::sync::{Arc, OnceLock};
4
5use liminal::channel::{ChannelSupervisor, InboxInstall};
6use liminal::durability::{DurableStore, bridge::block_on};
7use liminal::protocol::SchemaId;
8use serde_json::json;
9
10use super::{
11    binding,
12    config::AgentsConfig,
13    error::AgentError,
14    presence::Presence,
15    records::{AgentEvent, AgentTransition},
16    registry::{AgentRegistry, AgentSeats, REGISTRY_STREAM},
17    retained::Retained,
18};
19use crate::auth_lys::{HotAction, session::LysSession};
20use crate::config::types::ServerConfig;
21use crate::console::wall::WallBase;
22use crate::server::connection::services::{ConnectionSubscription, LiminalConnectionServices};
23use crate::services::lifecycle::Placer;
24
25/// Agent metadata never hosts a provider, harness pipe or process.
26#[derive(Debug)]
27pub struct Agents {
28    registry: Arc<AgentRegistry>,
29    retained: Arc<Retained>,
30    presence: Presence,
31    bounds: AgentsConfig,
32    network: String,
33    wall: WallBase,
34    listens: OnceLock<Arc<super::listens::Listens>>,
35}
36
37/// One subscription request with the connection's verified identity.
38pub struct AgentSubscription<'a> {
39    /// The server's ordinary channel adapter.
40    pub services: &'a LiminalConnectionServices,
41    /// The connection process id.
42    pub connection: u64,
43    /// The admitted pass session.
44    pub session: Option<&'a dyn LysSession>,
45    /// The exact listened subject.
46    pub subject: &'a str,
47    /// Schemas the client accepts.
48    pub schemas: &'a [SchemaId],
49    /// The connection's installed inbox bounds and notifier.
50    pub install: Option<InboxInstall>,
51    /// A server-allocated subscription id.
52    pub id: u64,
53}
54
55impl Agents {
56    /// Binds the console once; subscription resources never keep it alive.
57    ///
58    /// # Errors
59    /// A console was already installed.
60    pub fn bind_console(
61        &self,
62        console: &Arc<crate::console::ConsoleHub>,
63    ) -> Result<(), AgentError> {
64        self.listens
65            .set(super::listens::Listens::new(self.bounds, console))
66            .map_err(|_| AgentError::Invalid("agent console already bound"))?;
67        self.retained.bind_console(console)?;
68        Ok(())
69    }
70
71    /// Exact subjects of the live subscriptions, each once.
72    ///
73    /// # Errors
74    /// The listen index is unavailable.
75    pub fn listened_subjects(&self, connection: u64) -> Result<Vec<String>, AgentError> {
76        self.listens
77            .get()
78            .map_or(Ok(Vec::new()), |listens| listens.subjects(connection))
79    }
80
81    /// Wraps the real resource so unsubscribe, overflow and Drop remove its stop.
82    ///
83    /// # Errors
84    /// A listen bound or index failure.
85    pub fn track_listen(
86        &self,
87        connection: u64,
88        id: u64,
89        subject: &str,
90        resource: Box<dyn crate::server::connection::services::SubscriptionResource>,
91    ) -> Result<Box<dyn crate::server::connection::services::SubscriptionResource>, AgentError>
92    {
93        if self.presence.agent(connection)?.is_none() {
94            return Ok(resource);
95        }
96        self.listens
97            .get()
98            .ok_or(AgentError::Invalid("agent console is not bound"))?
99            .attach(connection, id, subject, resource)
100    }
101    /// Applies an agent draft pulled and authenticated by AGT-004, then seeds
102    /// the retained subjects. Replays reuse the committed mutation receipt.
103    ///
104    /// # Errors
105    /// Draft, bound, store or retained-subject failure.
106    pub fn execute_approved(
107        &self,
108        draft: &lys_pass::drafts::ApprovedDraft,
109    ) -> Result<lys_pass::drafts::Execution, AgentError> {
110        let receipt = self.registry.execute_approved_at(draft, self.now()?)?;
111        if draft.target.action == "create_agent" {
112            self.created(&draft.target.id)?;
113        }
114        Ok(receipt)
115    }
116    /// Opens over the server's own store and channel scheduler. Previously
117    /// present tickets are durably recorded gone before admission opens.
118    ///
119    /// # Errors
120    /// Invalid configuration, corrupt history, or a store or clock failure.
121    pub fn open(
122        store: Arc<dyn DurableStore>,
123        bounds: AgentsConfig,
124        network: String,
125        supervisor: ChannelSupervisor,
126        device: &str,
127        placer: Option<Arc<dyn Placer>>,
128        seats: Option<Arc<dyn AgentSeats>>,
129    ) -> Result<Arc<Self>, AgentError> {
130        let registry = AgentRegistry::open(store, bounds, network.clone(), placer, seats)?;
131        let retained = Retained::new(bounds, supervisor, device);
132        let wall = WallBase::now().map_err(|_| AgentError::Invalid("agent wall clock"))?;
133        let now = wall
134            .unix_ms_at(std::time::Instant::now())
135            .map_err(|_| AgentError::Invalid("agent wall clock"))?;
136        for record in registry.records()? {
137            if let Some(connection) = record.connection() {
138                registry.observe(
139                    &record.name,
140                    AgentTransition {
141                        holder: "server".into(),
142                        at_ms: now,
143                        change: AgentEvent::Gone {
144                            connection,
145                            reason: "server_restart".into(),
146                        },
147                    },
148                )?;
149            }
150            let current = registry.record(&record.name)?;
151            let last_gone =
152                current
153                    .transitions
154                    .iter()
155                    .rev()
156                    .find_map(|transition| match &transition.change {
157                        AgentEvent::Gone { reason, .. } => {
158                            Some((transition.at_ms, reason.as_str()))
159                        }
160                        _ => None,
161                    });
162            let (at_ms, reason) = last_gone.unwrap_or((current.created_at_ms, "created"));
163            retained.publish(&[
164                (
165                    format!("agents.{}.presence", record.name),
166                    json!({"present":false,"at_ms":at_ms,"reason":reason}),
167                ),
168                (
169                    format!("agents.{}.state", record.name),
170                    json!({"state":"gone"}),
171                ),
172            ])?;
173        }
174        Ok(Arc::new(Self {
175            registry,
176            retained,
177            presence: Presence::default(),
178            bounds,
179            network,
180            wall,
181            listens: OnceLock::new(),
182        }))
183    }
184
185    /// The one registry shared with the approved-draft executor.
186    #[must_use]
187    pub const fn registry(&self) -> &Arc<AgentRegistry> {
188        &self.registry
189    }
190    /// The one retained-subject writer shared with approvals.
191    #[must_use]
192    pub const fn retained(&self) -> &Arc<Retained> {
193        &self.retained
194    }
195    /// The configured positive bounds.
196    #[must_use]
197    pub const fn bounds(&self) -> &AgentsConfig {
198        &self.bounds
199    }
200    /// The network whose placed resources decide these subjects.
201    #[must_use]
202    pub fn network(&self) -> &str {
203        &self.network
204    }
205    /// The observed presence, never inferred from a lifecycle request.
206    #[must_use]
207    pub const fn presence(&self) -> &Presence {
208        &self.presence
209    }
210
211    /// Seeds subjects after a draft created a station. Existing presence is
212    /// preserved; a second executor notification cannot mark a live seat gone.
213    ///
214    /// # Errors
215    /// Unknown station or failed retained delivery.
216    pub fn created(&self, name: &str) -> Result<(), AgentError> {
217        let record = self.registry.record(name)?;
218        if self.presence.connection(name)?.is_none() {
219            self.retained.publish(&[
220                (
221                    format!("agents.{name}.presence"),
222                    json!({"present":false,"at_ms":record.created_at_ms,"reason":"created"}),
223                ),
224                (format!("agents.{name}.state"), json!({"state":"gone"})),
225            ])?;
226        }
227        Ok(())
228    }
229
230    /// Opens the real inbox or a retained metadata watcher after both identity
231    ///
232    /// and subject rights are decided. The caller's revocation and membership
233    /// checks still precede this method and still run at delivery release.
234    ///
235    /// # Errors
236    /// The pass, station, duplicate seat, schema or subscription is refused.
237    pub fn subscribe(
238        &self,
239        request: AgentSubscription<'_>,
240        ordinary: impl FnOnce(Option<InboxInstall>) -> Result<ConnectionSubscription, AgentError>,
241    ) -> Result<ConnectionSubscription, AgentError> {
242        let AgentSubscription {
243            services,
244            connection,
245            session,
246            subject,
247            schemas,
248            install,
249            id,
250        } = request;
251        let now = self.now()?;
252        binding::decide(session, &self.network, HotAction::Subscribe, subject, now)?;
253        let (name, leaf) = binding::subject(subject).ok_or(AgentError::Invalid("agent subject"))?;
254        if matches!(leaf, "presence" | "state") {
255            self.registry.record(name)?;
256            return self.retained.subscribe(id, subject, schemas, install);
257        }
258        if !self.registry.record(name)?.active() {
259            return Err(AgentError::Transition("agent is retired"));
260        }
261        if leaf != "inbox" {
262            return ordinary(install);
263        }
264        let session = session.ok_or(AgentError::Invalid("agent pass"))?;
265        let subscription = self.presence.connect(
266            &self.registry,
267            &self.retained,
268            super::presence::ConnectionObservation {
269                name,
270                holder: session.holder(),
271                connection,
272                at_ms: now,
273            },
274            || ordinary(install),
275        )?;
276        services
277            .console()
278            .emit(crate::console::TopologyChange::Service {
279                subject: subject.into(),
280                present: true,
281            })
282            .map_err(|_| AgentError::Delivery("agent topology change failed"))?;
283        Ok(subscription)
284    }
285
286    /// A transport's named close reaches the same durable observation path.
287    ///
288    /// # Errors
289    /// A store, clock or delivery failure; no automatic retry is added.
290    pub fn close(&self, connection: u64, reason: &str) -> Result<(), AgentError> {
291        self.bounds.field(reason, "transport reason")?;
292        if let Some(listens) = self.listens.get() {
293            listens.close(connection)?;
294        }
295        self.presence.close(
296            &self.registry,
297            &self.retained,
298            connection,
299            reason,
300            self.now()?,
301        )
302    }
303
304    pub(crate) fn now(&self) -> Result<u64, AgentError> {
305        self.wall
306            .unix_ms_at(std::time::Instant::now())
307            .map_err(|_| AgentError::Invalid("agent wall clock"))
308    }
309}
310
311/// Checks the durable namespace even when the configuration block is absent.
312/// An old installation with no agent rows keeps its existing behaviour.
313///
314/// # Errors
315/// A placed agent without `[auth.lys.agents]`, or a store read failure.
316pub fn configured(
317    config: &ServerConfig,
318    store: Arc<dyn DurableStore>,
319    supervisor: ChannelSupervisor,
320    device: &str,
321    runtime: Option<tokio::runtime::Handle>,
322) -> Result<Option<Arc<Agents>>, crate::ServerError> {
323    let lys = config.auth.as_ref().and_then(|auth| auth.lys.as_ref());
324    let bounds = lys.and_then(|lys| lys.agents);
325    let Some(bounds) = bounds else {
326        let head = block_on(store.read_stream_next_seq(REGISTRY_STREAM))
327            .map_err(|error| AgentError::Store(error.to_string()))?
328            .map_err(|error| AgentError::Store(error.to_string()))?;
329        if head.is_some() {
330            return Err(crate::ServerError::ConfigValidation {
331                message: "[auth.lys.agents] is required when a liminal.agent is placed".into(),
332            });
333        }
334        return Ok(None);
335    };
336    let lys = lys.ok_or(AgentError::Invalid("Lys configuration"))?;
337    let (placer, seats) = super::effects::configured(lys, runtime)?;
338    Ok(Some(Agents::open(
339        store,
340        bounds,
341        lys.network_id.clone(),
342        supervisor,
343        device,
344        placer,
345        seats,
346    )?))
347}
348
349/// Records a connection-fate close by the cause its class names: a client's
350/// `Disconnect` is `client_close`, as on the ordinary close path.
351pub fn connection_fate_closed(
352    services: &dyn crate::server::connection::services::ConnectionServices,
353    connection: u64,
354    class: crate::server::participant::ConnectionFateClass,
355) {
356    use crate::server::participant::ConnectionFateClass;
357    let reason = match class {
358        ConnectionFateClass::CleanDisconnect => "client_close",
359        ConnectionFateClass::ServerShutdown => "server_shutdown",
360        ConnectionFateClass::ConnectionLost => "transport_drop",
361        ConnectionFateClass::ProtocolError => "protocol_error",
362    };
363    connection_closed(services, connection, reason);
364}
365
366/// Records the transport's final stable cause. The caller supplies the cause
367///
368/// it actually observed; the Drop backstop uses `transport_drop`. No payload
369/// or credential is retained in a diagnostic if the durable write fails.
370pub fn connection_closed(
371    services: &dyn crate::server::connection::services::ConnectionServices,
372    connection: u64,
373    reason: &str,
374) {
375    if let Some(agents) = services.agents() {
376        if let Err(error) = agents.close(connection, reason) {
377            tracing::error!(connection, %error, "agent_presence_departure_failed");
378        }
379    }
380}
liminal_server/agents/runtime.rs

liminal_server/agents/
runtime.rs
