liminal_server/services/
site.rs

1//! Contracts-first construction, with immutable acts and atomic receipts.
2
3use std::os::unix::fs::PermissionsExt;
4use std::sync::Arc;
5
6use serde::{Deserialize, Serialize};
7
8use super::{Lifecycle, LifecycleError, ServiceRecord, ServiceState, Transition};
9use crate::config::templates::is_segment;
10
11#[path = "site/acts.rs"]
12mod acts;
13#[path = "site/effects.rs"]
14pub mod effects;
15#[path = "site/executor.rs"]
16pub mod executor;
17#[path = "site/outright.rs"]
18mod outright;
19#[path = "site/preparation.rs"]
20pub mod preparation;
21#[path = "site/records.rs"]
22mod records;
23#[path = "site/storage.rs"]
24pub mod storage;
25#[path = "site/topology.rs"]
26mod topology;
27pub use records::{
28    Approval, BuilderGrants, ConstructionConfig, Contract, SiteAct, SiteChange, SiteLimits,
29    SiteSpec,
30};
31
32/// All construction refusals contain boundary names, never rejected payloads.
33#[derive(Debug, thiserror::Error)]
34pub enum SiteError {
35    /// The boundary named here is missing or invalid.
36    #[error("site_invalid: {0}")]
37    Invalid(&'static str),
38    /// A configured count or byte bound was reached.
39    #[error("site_bound_exceeded: {0}")]
40    Bound(&'static str),
41    /// Safe business words generated from validated routing metadata.
42    #[error("{0}")]
43    Refused(String),
44    /// Admission is poisoned after an uncertain durable operation.
45    #[error("site_store_unavailable")]
46    Store,
47    /// The admitted pass's own refusal, without token or payload bytes.
48    #[error("site_pass_refused: {0}")]
49    Pass(String),
50    /// The existing lifecycle failed.
51    #[error("{0}")]
52    Lifecycle(#[from] LifecycleError),
53}
54
55/// The schema strings are carried exactly; equivalence compares parsed shapes.
56///
57/// # Errors
58/// Missing or invalid schema, reserved subject, or configured boundary refusal.
59pub fn validate_contracts(spec: &SiteSpec, limits: &SiteLimits) -> Result<(), SiteError> {
60    if !is_segment(&spec.name) || spec.name.len() > limits.text_bytes.get() {
61        return Err(SiteError::Invalid("name"));
62    }
63    if spec.purpose.trim().is_empty()
64        || spec.purpose.contains(['\n', '\r'])
65        || spec.purpose.len() > limits.text_bytes.get()
66    {
67        return Err(SiteError::Invalid("purpose: one sentence"));
68    }
69    if spec.subjects.is_empty() || spec.subjects.len() > limits.subjects.get() {
70        return Err(SiteError::Bound("subjects"));
71    }
72    let mut seen = std::collections::BTreeSet::new();
73    for subject in &spec.subjects {
74        if subject.len() > limits.text_bytes.get()
75            || !subject.split('.').all(is_segment)
76            || subject.starts_with("console.")
77            || subject.starts_with("agents.")
78            || subject.starts_with("services.")
79            || crate::server::connection::services::is_site_route(subject)
80            || !seen.insert(subject)
81        {
82            return Err(SiteError::Invalid("subject"));
83        }
84        let contract = spec
85            .contracts
86            .get(subject)
87            .ok_or_else(|| SiteError::Refused(format!("subject {subject} has no schema")))?;
88        if contract.input_schema_json.is_empty() || contract.output_schema_json.is_empty() {
89            return Err(SiteError::Refused(format!(
90                "subject {subject} has no schema"
91            )));
92        }
93        for (kind, bytes) in [
94            ("input", &contract.input_schema_json),
95            ("output", &contract.output_schema_json),
96        ] {
97            if bytes.len() > limits.schema_bytes.get() {
98                return Err(SiteError::Bound("schema_bytes"));
99            }
100            let value = serde_json::from_str(bytes).map_err(|_| {
101                SiteError::Refused(format!("subject {subject} has no valid {kind} schema"))
102            })?;
103            liminal::channel::Schema::new(value).map_err(|_| {
104                SiteError::Refused(format!("subject {subject} has no valid {kind} schema"))
105            })?;
106        }
107    }
108    if spec.contracts.len() != seen.len() {
109        return Err(SiteError::Invalid("contracts name undeclared subjects"));
110    }
111    Ok(())
112}
113
114/// Creates `dir` private to the server (mode 0700) when it does not exist,
115/// whatever the process umask; an existing directory keeps its mode and is
116/// judged by the check that follows, since it is the operator's (SRV-036).
117pub(super) fn create_private_dir(dir: &std::path::Path) -> std::io::Result<()> {
118    use std::os::unix::fs::DirBuilderExt;
119    if dir.try_exists()? {
120        return Ok(());
121    }
122    if let Some(parent) = dir.parent() {
123        std::fs::create_dir_all(parent)?;
124    }
125    std::fs::DirBuilder::new().mode(0o700).create(dir)
126}
127
128impl Lifecycle {
129    /// Opens construction with explicit migration bounds before parsing any old
130    /// file. Ordinary configuration-only registries retain their existing path.
131    ///
132    /// # Errors
133    /// Invalid old snapshot, nonregular file, count/byte bound, or migration write.
134    pub fn open_bounded(
135        templates: &[crate::config::types::ServiceTemplateConfig],
136        dir: std::path::PathBuf,
137        network: String,
138        placer: Option<Arc<dyn super::Placer>>,
139        limits: &SiteLimits,
140    ) -> Result<Arc<Self>, LifecycleError> {
141        create_private_dir(&dir).map_err(|error| super::store(&dir, &error))?;
142        if std::fs::metadata(&dir)
143            .map_err(|error| super::store(&dir, &error))?
144            .permissions()
145            .mode()
146            & 0o022
147            != 0
148        {
149            return Err(LifecycleError::Store(
150                "construction_registry_writable_by_others".into(),
151            ));
152        }
153        let snapshot = storage::load_bounded(&dir, limits)?;
154        Ok(Self::open_snapshot(
155            templates, dir, network, placer, snapshot,
156        ))
157    }
158
159    /// Installs operator bounds and the grant effect once, before admission.
160    /// The retained projection shares the existing scheduler and inbox bounds.
161    ///
162    /// # Errors
163    /// Repeated installation or invalid restored construction records.
164    pub fn configure_sites(
165        &self,
166        limits: SiteLimits,
167        store: Arc<dyn liminal::durability::DurableStore>,
168        templates: Option<super::templates::TemplatePolicy>,
169        effects: Arc<dyn effects::SiteEffects>,
170        retained: Arc<crate::agents::retained::Retained>,
171    ) -> Result<(), SiteError> {
172        if std::fs::metadata(&self.dir)
173            .map_err(|_| SiteError::Store)?
174            .permissions()
175            .mode()
176            & 0o022
177            != 0
178        {
179            return Err(SiteError::Invalid(
180                "construction registry must not be writable by group or others",
181            ));
182        }
183        self.site_config
184            .set(storage::SiteBindings {
185                store,
186                limits,
187                templates,
188                effects,
189                retained,
190            })
191            .map_err(|_| SiteError::Invalid("construction already configured"))?;
192        let mut registry = self.lock()?;
193        storage::restore(self, &mut registry)?;
194        drop(registry);
195        self.validate_site_storage()?;
196        let registry = self.lock()?;
197        for record in registry
198            .records
199            .values()
200            .filter(|record| record.purpose.is_some())
201        {
202            validate_contracts(
203                &SiteSpec {
204                    name: record.name.clone(),
205                    purpose: record.purpose.clone().ok_or(SiteError::Store)?,
206                    subjects: record.subjects.clone(),
207                    contracts: record.contracts.clone(),
208                },
209                &self.site_bindings()?.limits,
210            )?;
211            self.publish_plan(record)?;
212        }
213        drop(registry);
214        Ok(())
215    }
216
217    pub(super) fn site_bindings(&self) -> Result<&storage::SiteBindings, SiteError> {
218        self.site_config.get().ok_or(SiteError::Invalid(
219            "[auth.lys.construction] is not configured",
220        ))
221    }
222
223    pub(super) fn building(&self, id: &str) -> Result<ServiceRecord, SiteError> {
224        let record = self.record(id)?;
225        if record.state != ServiceState::Building {
226            return Err(SiteError::Refused(format!(
227                "service {} is not building",
228                record.name
229            )));
230        }
231        Ok(record)
232    }
233
234    pub(super) fn publish_plan(&self, record: &ServiceRecord) -> Result<(), SiteError> {
235        let retained = &self.site_bindings()?.retained;
236        let subject = format!("services.{}.plan", record.name);
237        let value = super::plan::current_value(&record.plan);
238        if retained
239            .value(&subject)
240            .map_err(|_| SiteError::Store)?
241            .as_ref()
242            == Some(&value)
243        {
244            return Ok(());
245        }
246        retained
247            .publish(&[(subject, value)])
248            .map_err(|_| SiteError::Store)
249    }
250
251    pub(super) fn site_transition(
252        record: &mut ServiceRecord,
253        holder: &str,
254        at_ms: u64,
255        words: &str,
256    ) {
257        record.transitions.push(Transition {
258            from: record.state,
259            to: record.state,
260            holder: holder.into(),
261            at_ms,
262            note: Some(words.into()),
263        });
264    }
265}
266
267/// A construction snapshot never exposes its free text through Debug.
268#[derive(Clone, Default, Serialize, Deserialize)]
269#[serde(deny_unknown_fields)]
270pub(super) struct Construction {
271    pub templates: std::collections::BTreeMap<String, super::templates::TemplateRecord>,
272    pub drafts: std::collections::BTreeMap<String, executor::Prepared>,
273    pub outcomes: std::collections::BTreeMap<String, executor::Outcome>,
274}
275
276#[cfg(test)]
277#[path = "site/test_support.rs"]
278pub(crate) mod test_support;
279#[cfg(test)]
280#[path = "site_tests.rs"]
281mod tests;
282
283impl Lifecycle {
284    /// Network placement used for the caller's exact permission binding.
285    #[must_use]
286    pub fn network_id(&self) -> &str {
287        &self.network
288    }
289    /// Returns the explicitly configured positive construction limits.
290    ///
291    /// # Errors
292    /// Construction is not configured.
293    pub fn construction_limits(&self) -> Result<SiteLimits, SiteError> {
294        Ok(self.site_bindings()?.limits.clone())
295    }
296    /// Subscribes to the current plan before any later changes, under its writer lock.
297    ///
298    /// # Errors
299    /// Missing subject, watcher bound, schema or inbox refusal.
300    pub fn subscribe_plan(
301        &self,
302        id: u64,
303        subject: &str,
304        schemas: &[liminal::protocol::SchemaId],
305        install: Option<liminal::channel::InboxInstall>,
306    ) -> Result<crate::server::connection::services::ConnectionSubscription, SiteError> {
307        self.site_bindings()?
308            .retained
309            .subscribe(id, subject, schemas, install)
310            .map_err(|error| SiteError::Refused(error.to_string()))
311    }
312}
313
314impl Lifecycle {
315    /// Installs the existing hub's event signal without another publisher or timer.
316    ///
317    /// # Errors
318    /// The signal was already bound.
319    pub fn bind_site_console(
320        &self,
321        console: &Arc<crate::console::ConsoleHub>,
322    ) -> Result<(), SiteError> {
323        self.site_console
324            .set(Arc::downgrade(console))
325            .map_err(|_| SiteError::Invalid("site console already bound"))
326    }
327    /// Repairs retained current values after a durable outcome or restart.
328    ///
329    /// # Errors
330    /// A target, retained stream or hub signal failed.
331    pub fn refresh_sites(&self) -> Result<(), SiteError> {
332        let registry = self.lock()?;
333        for record in registry
334            .records
335            .values()
336            .filter(|record| record.purpose.is_some())
337        {
338            self.publish_plan(record)?;
339        }
340        drop(registry);
341        self.notify_sites()
342    }
343    pub(super) fn notify_sites(&self) -> Result<(), SiteError> {
344        if let Some(console) = self.site_console.get().and_then(std::sync::Weak::upgrade) {
345            console
346                .agent_changed()
347                .map_err(|_| SiteError::Invalid("site topology signal failed"))?;
348        }
349        Ok(())
350    }
351}
liminal_server/services/site.rs

liminal_server/services/
site.rs
