liminal_server/agents/
registry.rs

1//! One atomic event batch and one flush per transition; no history is pruned.
2
3use std::sync::{Arc, Mutex, MutexGuard};
4
5use liminal::durability::{DurableStore, bridge::block_on};
6use lys_pass::drafts::{ApprovedDraft, Execution, request_digest};
7
8use super::config::AgentsConfig;
9use super::error::AgentError;
10use super::records::{AgentAct, AgentEvent, AgentRecord, AgentTransition};
11
12#[path = "self_station.rs"]
13mod self_station;
14#[path = "registry_storage.rs"]
15mod storage;
16use crate::services::lifecycle::Placer;
17use storage::{
18    Index, Land, Receipt, action, apply, check_bytes, check_population, refusal_name, store_error,
19    validate_act,
20};
21
22/// A new namespace leaves an old installation's service records unchanged.
23pub const REGISTRY_STREAM: &str = "liminal/agents/v1";
24
25/// Lys owns every process; this seam cannot execute a program locally.
26pub trait AgentSeats: Send + Sync {
27    /// Starts a reviewed seat once under `operation`.
28    ///
29    /// # Errors
30    /// Lys's named refusal, without a response body.
31    fn start(&self, name: &str, operation: &str) -> Result<(), AgentError>;
32    /// Stops that seat once under `operation`.
33    ///
34    /// # Errors
35    /// Lys's named refusal, without a response body.
36    fn stop(&self, name: &str, operation: &str) -> Result<(), AgentError>;
37}
38
39/// Registry over the server's own durable store. No pass or message is retained.
40pub struct AgentRegistry {
41    store: Arc<dyn DurableStore>,
42    bounds: AgentsConfig,
43    network: String,
44    placer: Option<Arc<dyn Placer>>,
45    seats: Option<Arc<dyn AgentSeats>>,
46    index: Mutex<Index>,
47}
48
49impl std::fmt::Debug for AgentRegistry {
50    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51        out.debug_struct("AgentRegistry")
52            .field("bounds", &self.bounds)
53            .finish_non_exhaustive()
54    }
55}
56
57impl AgentRegistry {
58    /// Restores every transition in bounded pages, including retired agents.
59    /// `placer` and `seats` are production Lys bindings, never authority to
60    /// approve a draft. Only the authenticated approved-draft connector may
61    /// call [`Self::execute_approved`].
62    ///
63    /// # Errors
64    /// Invalid bounds, corrupt history, or a failed store read.
65    pub fn open(
66        store: Arc<dyn DurableStore>,
67        bounds: AgentsConfig,
68        network: String,
69        placer: Option<Arc<dyn Placer>>,
70        seats: Option<Arc<dyn AgentSeats>>,
71    ) -> Result<Arc<Self>, AgentError> {
72        bounds.validate()?;
73        bounds.field(&network, "network")?;
74        let mut index = Index::default();
75        loop {
76            let page =
77                block_on(store.read_from(REGISTRY_STREAM, index.next, bounds.replay_page_rows))
78                    .map_err(store_error)?
79                    .map_err(store_error)?;
80            if page.is_empty() {
81                break;
82            }
83            for row in page {
84                if row.sequence != index.next {
85                    return Err(AgentError::Invalid("history sequence"));
86                }
87                check_bytes(&bounds, row.payload.len())?;
88                let land: Land = serde_json::from_slice(&row.payload)
89                    .map_err(|_| AgentError::Invalid("stored agent event"))?;
90                if land.version != 1 {
91                    return Err(AgentError::Invalid("stored event version"));
92                }
93                validate_act(&bounds, &land.name, &land.transition)?;
94                bounds.field(&land.owner, "owner")?;
95                apply(&mut index, land)?;
96            }
97        }
98        Ok(Arc::new(Self {
99            store,
100            bounds,
101            network,
102            placer,
103            seats,
104            index: Mutex::new(index),
105        }))
106    }
107
108    fn lock(&self) -> Result<MutexGuard<'_, Index>, AgentError> {
109        let index = self
110            .index
111            .lock()
112            .map_err(|_| AgentError::Store("lock poisoned".into()))?;
113        if index.uncertain {
114            return Err(AgentError::Uncertain);
115        }
116        Ok(index)
117    }
118
119    /// Reads one whole history, without inventing a transition after restart.
120    ///
121    /// # Errors
122    /// The name is unknown or the store outcome is unresolved.
123    pub fn record(&self, name: &str) -> Result<AgentRecord, AgentError> {
124        self.lock()?
125            .records
126            .get(name)
127            .cloned()
128            .ok_or_else(|| AgentError::Unknown(name.into()))
129    }
130
131    /// Every record, including retired stations, in name order.
132    ///
133    /// # Errors
134    /// The index is unavailable.
135    pub fn records(&self) -> Result<Vec<AgentRecord>, AgentError> {
136        Ok(self.lock()?.records.values().cloned().collect())
137    }
138
139    /// Compact station metadata; topology never clones a transition history.
140    ///
141    /// # Errors
142    /// The index is unavailable.
143    pub fn station_records(&self) -> Result<Vec<AgentRecord>, AgentError> {
144        Ok(self
145            .lock()?
146            .records
147            .values()
148            .map(|record| AgentRecord {
149                name: record.name.clone(),
150                harness: record.harness,
151                purpose: record.purpose.clone(),
152                owner: record.owner.clone(),
153                service: record.service.clone(),
154                created_at_ms: record.created_at_ms,
155                transitions: record.transitions.last().cloned().into_iter().collect(),
156            })
157            .collect())
158    }
159
160    /// Reads the receipt committed in the same batch as its mutation.
161    ///
162    /// # Errors
163    /// The index is unavailable.
164    pub fn receipt(&self, draft: &str) -> Result<Option<Execution>, AgentError> {
165        Ok(self.lock()?.receipts.get(draft).cloned())
166    }
167
168    /// A held response cannot mutate the registry.
169    #[must_use]
170    pub fn held(&self, action: &str, draft: &str) -> AgentError {
171        AgentError::Held {
172            action: action.into(),
173            draft: draft.into(),
174        }
175    }
176
177    /// Executes only a draft pulled from Lys by the authenticated connector.
178    /// Its target and immutable digest are checked again here. Receipt and
179    /// transition share one event, so a crash cannot keep one without the other.
180    /// Lys starts and stops use the draft id as their idempotency operation.
181    ///
182    /// # Errors
183    /// Foreign or changed drafts, a bound, Lys's refusal, or an unresolved write.
184    pub fn execute_approved(&self, draft: &ApprovedDraft) -> Result<Execution, AgentError> {
185        let at_ms = u64::try_from(
186            std::time::SystemTime::now()
187                .duration_since(std::time::UNIX_EPOCH)
188                .map_err(|_| AgentError::Invalid("agent execution clock"))?
189                .as_millis(),
190        )
191        .map_err(|_| AgentError::Invalid("agent execution clock overflow"))?;
192        self.execute_approved_at(draft, at_ms)
193    }
194
195    /// Uses the execution owner's observed time, not the draft's preparation time.
196    ///
197    /// # Errors
198    /// The same refusals as [`Self::execute_approved`].
199    pub fn execute_approved_at(
200        &self,
201        draft: &ApprovedDraft,
202        at_ms: u64,
203    ) -> Result<Execution, AgentError> {
204        self.bounds.field(&draft.id, "draft")?;
205        self.bounds.field(&draft.grant, "grant")?;
206        check_bytes(&self.bounds, draft.words.len())?;
207        if draft.app != "liminal"
208            || draft.target.kind != "liminal.agent"
209            || request_digest(&draft.words) != draft.request_digest
210        {
211            return Err(AgentError::Invalid("approved draft target or digest"));
212        }
213        let act: AgentAct = serde_json::from_str(&draft.words)
214            .map_err(|_| AgentError::Invalid("approved agent act"))?;
215        if act.name != draft.target.id || action(&act.change) != Some(draft.target.action.as_str())
216        {
217            return Err(AgentError::Invalid("approved agent action"));
218        }
219        let transition = AgentTransition {
220            holder: act.holder,
221            at_ms,
222            change: act.change,
223        };
224        validate_act(&self.bounds, &act.name, &transition)?;
225        self.bounds.field(&act.owner, "owner")?;
226        let mut index = self.lock()?;
227        if let Some(receipt) = index.receipts.get(&draft.id) {
228            return match receipt {
229                Execution::Executed { request_digest, .. }
230                | Execution::RefusedOnExecution { request_digest, .. }
231                    if request_digest == &draft.request_digest =>
232                {
233                    Ok(receipt.clone())
234                }
235                _ => Err(AgentError::DraftMismatch),
236            };
237        }
238        Self::prepare(&index, &act.name, &transition.change)?;
239        if !matches!(transition.change, AgentEvent::Create(_))
240            && index
241                .records
242                .get(&act.name)
243                .is_some_and(|record| record.owner != act.owner)
244        {
245            return Err(AgentError::Invalid(
246                "agent owner differs from its prepared act",
247            ));
248        }
249        if matches!(transition.change, AgentEvent::Create(_)) {
250            check_population(&index, &self.bounds, &act.owner)?;
251        }
252        let receipt = Execution::Executed {
253            request_digest: draft.request_digest.clone(),
254            receipt_digest: request_digest(&format!(
255                "{}:{}:{}",
256                draft.id, draft.request_digest, index.next
257            )),
258        };
259        let land = Land {
260            version: 1,
261            name: act.name.clone(),
262            owner: act.owner,
263            transition: transition.clone(),
264            receipt: Some(Receipt {
265                draft: draft.id.clone(),
266                execution: receipt.clone(),
267            }),
268        };
269        let prepared =
270            serde_json::to_vec(&land).map_err(|_| AgentError::Invalid("event encoding"))?;
271        check_bytes(&self.bounds, prepared.len())?;
272        match &transition.change {
273            AgentEvent::Create(_) => self.place(&act.name, &draft.id)?,
274            AgentEvent::Start => self
275                .seats
276                .as_ref()
277                .ok_or(AgentError::Invalid("Lys seat binding is not configured"))?
278                .start(&act.name, &draft.id)?,
279            AgentEvent::Stop => self
280                .seats
281                .as_ref()
282                .ok_or(AgentError::Invalid("Lys seat binding is not configured"))?
283                .stop(&act.name, &draft.id)?,
284            AgentEvent::Retire => {
285                if index
286                    .records
287                    .get(&act.name)
288                    .is_some_and(|record| record.connection().is_some())
289                {
290                    return Err(AgentError::Transition("stop and close before retirement"));
291                }
292            }
293            _ => return Err(AgentError::Invalid("presence is observed by the server")),
294        }
295        self.land(&mut index, land)?;
296        Ok(receipt)
297    }
298
299    /// Records a connection observation after identity and rights were decided.
300    ///
301    /// # Errors
302    /// Not a presence event, a stale connection, or a store failure.
303    pub(crate) fn observe(
304        &self,
305        name: &str,
306        transition: AgentTransition,
307    ) -> Result<(), AgentError> {
308        if !matches!(
309            transition.change,
310            AgentEvent::Present { .. }
311                | AgentEvent::Gone { .. }
312                | AgentEvent::Session { .. }
313                | AgentEvent::SessionClosed { .. }
314        ) {
315            return Err(AgentError::Invalid("server observation"));
316        }
317        validate_act(&self.bounds, name, &transition)?;
318        let mut index = self.lock()?;
319        Self::prepare(&index, name, &transition.change)?;
320        let owner = index
321            .records
322            .get(name)
323            .ok_or_else(|| AgentError::Unknown(name.into()))?
324            .owner
325            .clone();
326        let result = self.land(
327            &mut index,
328            Land {
329                version: 1,
330                name: name.into(),
331                owner,
332                transition,
333                receipt: None,
334            },
335        );
336        drop(index);
337        result
338    }
339
340    fn prepare(index: &Index, name: &str, change: &AgentEvent) -> Result<(), AgentError> {
341        if let AgentEvent::Create(_) = change {
342            if index.records.contains_key(name) {
343                return Err(AgentError::NameTaken(name.into()));
344            }
345        } else {
346            let record = index
347                .records
348                .get(name)
349                .ok_or_else(|| AgentError::Unknown(name.into()))?;
350            if !record.active() {
351                return Err(AgentError::Transition("agent is retired"));
352            }
353            if let AgentEvent::Present { connection } = change {
354                if let Some((existing, since_ms)) = record.seat() {
355                    return Err(AgentError::Present {
356                        agent: name.into(),
357                        connection: existing,
358                        since_ms,
359                    });
360                }
361                if *connection == 0 {
362                    return Err(AgentError::Invalid("connection"));
363                }
364            }
365            if let AgentEvent::Gone { connection, .. } = change {
366                if record.connection() != Some(*connection) {
367                    return Err(AgentError::Transition("departure names another connection"));
368                }
369            }
370        }
371        Ok(())
372    }
373
374    fn place(&self, name: &str, draft: &str) -> Result<(), AgentError> {
375        let placer = self
376            .placer
377            .as_ref()
378            .ok_or(AgentError::Invalid("agent placer is not configured"))?;
379        let subjects = [
380            format!("agents.{name}.inbox"),
381            format!("agents.{name}.presence"),
382            format!("agents.{name}.state"),
383        ];
384        for child in std::iter::once(("liminal.agent", name)).chain(
385            subjects
386                .iter()
387                .map(|subject| ("liminal.subject", subject.as_str())),
388        ) {
389            // Lys's operation id is `op-` and exactly 32 hex digits.
390            let digest =
391                blake3::hash(format!("agent-place:{draft}:{}:{}", child.0, child.1).as_bytes());
392            let operation = format!("op-{}", &digest.to_hex()[..32]);
393            placer
394                .place(&operation, child, ("liminal.network", &self.network))
395                .map_err(|refusal| AgentError::Store(refusal_name(&refusal)))?;
396        }
397        Ok(())
398    }
399
400    fn land(&self, index: &mut Index, land: Land) -> Result<(), AgentError> {
401        if let AgentEvent::Create(_) = &land.transition.change {
402            check_population(index, &self.bounds, &land.owner)?;
403        }
404        let bytes = serde_json::to_vec(&land).map_err(|_| AgentError::Invalid("event encoding"))?;
405        check_bytes(&self.bounds, bytes.len())?;
406        let appended = block_on(
407            self.store
408                .append_all(REGISTRY_STREAM, vec![bytes], index.next),
409        )
410        .map_err(store_error)
411        .and_then(|answer| answer.map_err(store_error));
412        match appended {
413            Ok(next) if Some(next) == index.next.checked_add(1) => {}
414            other => {
415                tracing::warn!(expected = index.next, answer = ?other.map_err(|error| error.to_string()), "agent registry append outcome unknown");
416                index.uncertain = true;
417                return Err(AgentError::Uncertain);
418            }
419        }
420        if block_on(self.store.flush())
421            .map_err(store_error)
422            .and_then(|answer| answer.map_err(store_error))
423            .is_err()
424        {
425            index.uncertain = true;
426            return Err(AgentError::Uncertain);
427        }
428        apply(index, land)
429    }
430}
liminal_server/agents/registry.rs

liminal_server/agents/
registry.rs
