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

liminal_server/agents/
registry.rs
