systemprompt_loader/bundle/bootstrap/
mod.rs

1//! Boot-time resolution of the services root from configured bundle sources.
2//!
3//! The happy path is fetch, verify, compose, load-validate, swap, record. Every
4//! other path is a named fallback carrying the error that caused it: an
5//! instance serving last-good content reports [`ServicesProvenance::LastGood`]
6//! with the failure text, so "we are running yesterday's bundle" is visible
7//! rather than inferred from a log line that scrolled past. `fail_closed`
8//! refuses to boot instead.
9//!
10//! Copyright (c) systemprompt.io — Business Source License 1.1.
11//! See <https://systemprompt.io> for licensing details.
12
13pub mod baked;
14mod fetch;
15
16use std::collections::BTreeMap;
17use std::path::{Path, PathBuf};
18use std::time::Duration;
19
20use systemprompt_manifest::profile::{FetchFailurePolicy, Profile};
21use systemprompt_manifest::services::bundle::ServicesBundleState;
22
23use super::cache::BundleCache;
24use super::compose::{BundleMember, compose};
25use super::degraded::DegradedRecord;
26use super::error::{BundleError, BundleResult};
27use systemprompt_manifest::services::root::{ActiveServicesRoot, ServicesProvenance};
28
29use baked::{is_base, stage_baked_base};
30use fetch::{ResolvedSource, SourceContext, resolve_source};
31
32const HTTP_TIMEOUT: Duration = Duration::from_secs(60);
33const CONFIG_RELPATH: &str = "config/config.yaml";
34
35#[derive(Debug, Clone, Copy)]
36pub struct ServicesSourceBootstrap;
37
38impl ServicesSourceBootstrap {
39    pub async fn resolve(
40        profile: &Profile,
41        resolve_secret: impl Fn(&str) -> Option<String> + Send + Sync,
42        core_version: &str,
43    ) -> BundleResult<ActiveServicesRoot> {
44        if profile.services.sources.is_empty() {
45            let path = PathBuf::from(&profile.paths.services);
46            return Ok(ActiveServicesRoot {
47                base: path.clone(),
48                path,
49                provenance: ServicesProvenance::Bundled,
50            });
51        }
52
53        let cache = BundleCache::new(cache_root(profile));
54        if profile.server.role.reads_shared_services_composition() {
55            return shared_composition(profile, &cache);
56        }
57        match Self::compose_sources(profile, &cache, &resolve_secret, core_version).await {
58            Ok(active) => {
59                if let Err(error) = cache.clear_degraded() {
60                    tracing::warn!(error = %error, "Could not clear the degraded services record");
61                }
62                Ok(active)
63            },
64            Err(e) => {
65                tracing::error!(error = %e, "Services bundle refresh failed");
66                let active = Self::fall_back(profile, &cache, e)?;
67                if let Some(record) = DegradedRecord::for_provenance(&active.provenance)
68                    && let Err(error) = cache.write_degraded(&record)
69                {
70                    tracing::warn!(error = %error, "Could not record the degraded services source");
71                }
72                Ok(active)
73            },
74        }
75    }
76
77    async fn compose_sources(
78        profile: &Profile,
79        cache: &BundleCache,
80        resolve_secret: &(impl Fn(&str) -> Option<String> + Send + Sync),
81        core_version: &str,
82    ) -> BundleResult<ActiveServicesRoot> {
83        let client = reqwest::Client::builder()
84            .redirect(reqwest::redirect::Policy::none())
85            .timeout(HTTP_TIMEOUT)
86            .build()
87            .map_err(|e| BundleError::policy_context("http client", e))?;
88
89        let previous = cache.read_state();
90        let ctx = SourceContext {
91            cache,
92            state: &previous,
93            core_version,
94            client: &client,
95        };
96        let mut resolved: Vec<ResolvedSource> = Vec::new();
97        for source in &profile.services.sources {
98            let auth = source.auth_secret().and_then(resolve_secret);
99            resolved.push(resolve_source(source, &ctx, auth).await?);
100        }
101        // Why: a profile that pins only kits names no base; the tree baked
102        // into the image is the base, and it must be member zero so ownership
103        // and the authz reconcile treat every pinned source as a kit.
104        if !resolved
105            .first()
106            .is_some_and(|r| is_base(&r.signed.manifest))
107        {
108            let base = stage_baked_base(cache, Path::new(&profile.paths.services), core_version)?;
109            resolved.insert(0, base);
110        }
111
112        let members: Vec<BundleMember<'_>> = resolved
113            .iter()
114            .map(|r| BundleMember {
115                name: r.name.clone(),
116                content_hash: r.content_hash.clone(),
117                manifest: &r.signed.manifest,
118            })
119            .collect();
120        let (composed_path, composed_hash) = compose(cache, &members)?;
121        let kits: Vec<&str> = members.iter().map(|m| m.name.as_str()).collect();
122        let retained = previous.composed_hash == composed_hash;
123        refuse_unloadable(&composed_path, profile, &kits, retained)?;
124        cache.swap_current(&composed_path)?;
125
126        let state = ServicesBundleState {
127            composed_hash: composed_hash.clone(),
128            last_reconciled_hash: previous.last_reconciled_hash.clone(),
129            sources: resolved
130                .iter()
131                .map(|r| (r.name.clone(), r.state.clone()))
132                .collect(),
133        };
134        cache.write_state(&state)?;
135        cache.gc(&composed_hash)?;
136
137        let versions: BTreeMap<String, String> = resolved
138            .iter()
139            .map(|r| (r.name.clone(), r.signed.manifest.version.clone()))
140            .collect();
141        tracing::info!(composed_hash = %composed_hash, sources = resolved.len(), "Services bundles composed");
142
143        Ok(ActiveServicesRoot {
144            path: cache.current_link(),
145            base: PathBuf::from(&profile.paths.services),
146            provenance: ServicesProvenance::Fetched {
147                composed_hash,
148                versions,
149            },
150        })
151    }
152
153    fn fall_back(
154        profile: &Profile,
155        cache: &BundleCache,
156        error: BundleError,
157    ) -> BundleResult<ActiveServicesRoot> {
158        match profile.services.on_fetch_failure {
159            FetchFailurePolicy::FailClosed => Err(BundleError::policy_context(
160                "services.on_fetch_failure is fail_closed",
161                error,
162            )),
163            FetchFailurePolicy::UseLastGood => last_good(profile, cache, &error)
164                .map_or_else(|| bundled_fallback(profile, error), Ok),
165            FetchFailurePolicy::UseBundled => bundled_fallback(profile, error),
166        }
167    }
168}
169
170fn refuse_unloadable(
171    composed: &Path,
172    profile: &Profile,
173    kits: &[&str],
174    retained: bool,
175) -> BundleResult<()> {
176    let Err(error) =
177        crate::ConfigLoader::load_from_path(&composed.join(CONFIG_RELPATH), Some(profile))
178    else {
179        return Ok(());
180    };
181    if !retained && let Err(remove) = std::fs::remove_dir_all(composed) {
182        tracing::warn!(error = %remove, path = %composed.display(), "Could not remove the rejected composition");
183    }
184    Err(BundleError::policy_context(
185        format!(
186            "the composed services tree (sources: {}) does not load, so it was not swapped in",
187            kits.join(", ")
188        ),
189        error,
190    ))
191}
192
193fn shared_composition(profile: &Profile, cache: &BundleCache) -> BundleResult<ActiveServicesRoot> {
194    let state = cache.read_state();
195    let Some(current) = cache
196        .current_root()
197        .filter(|_| !state.composed_hash.is_empty())
198    else {
199        return Err(BundleError::NoSharedComposition {
200            role: profile.server.role.as_str(),
201            cache: cache_root(profile),
202        });
203    };
204    let versions = state
205        .sources
206        .iter()
207        .map(|(name, source)| (name.clone(), source.version.clone()))
208        .collect();
209    Ok(ActiveServicesRoot {
210        path: current,
211        base: PathBuf::from(&profile.paths.services),
212        provenance: ServicesProvenance::Fetched {
213            composed_hash: state.composed_hash,
214            versions,
215        },
216    })
217}
218
219fn last_good(
220    profile: &Profile,
221    cache: &BundleCache,
222    error: &BundleError,
223) -> Option<ActiveServicesRoot> {
224    let current = cache.current_root()?;
225    let state = cache.read_state();
226    if state.composed_hash.is_empty() {
227        return None;
228    }
229    tracing::error!(
230        composed_hash = %state.composed_hash,
231        error = %error,
232        "Serving the last-good services composition after a failed refresh"
233    );
234    Some(ActiveServicesRoot {
235        path: current,
236        base: PathBuf::from(&profile.paths.services),
237        provenance: ServicesProvenance::LastGood {
238            composed_hash: state.composed_hash,
239            error: error.to_string(),
240        },
241    })
242}
243
244fn bundled_fallback(profile: &Profile, error: BundleError) -> BundleResult<ActiveServicesRoot> {
245    let root = PathBuf::from(&profile.paths.services);
246    if !root.join("config/config.yaml").is_file() {
247        return Err(BundleError::policy_context(
248            format!(
249                "no cached bundle and no baked services tree at {}",
250                root.display()
251            ),
252            error,
253        ));
254    }
255    tracing::error!(
256        path = %root.display(),
257        error = %error,
258        "Serving the services tree baked into the image after a failed refresh"
259    );
260    Ok(ActiveServicesRoot {
261        base: root.clone(),
262        path: root,
263        provenance: ServicesProvenance::BundledFallback {
264            error: error.to_string(),
265        },
266    })
267}
268
269#[must_use]
270pub fn cache_root(profile: &Profile) -> PathBuf {
271    profile.services.cache_dir.as_ref().map_or_else(
272        || PathBuf::from(&profile.paths.system).join("services-cache"),
273        PathBuf::from,
274    )
275}
systemprompt_loader/bundle/bootstrap/mod.rs

systemprompt_loader/bundle/bootstrap/
mod.rs
