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, swap, record. Every other path
4//! is a named fallback carrying the error that caused it: an instance serving
5//! last-good content reports [`ServicesProvenance::LastGood`] with the failure
6//! text, so "we are running yesterday's bundle" is visible rather than
7//! inferred from a log line that scrolled past. `fail_closed` refuses to boot
8//! 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::PathBuf;
18use std::time::Duration;
19
20use systemprompt_models::profile::{FetchFailurePolicy, Profile};
21use systemprompt_models::services::bundle::ServicesBundleState;
22
23use super::cache::BundleCache;
24use super::compose::{BundleMember, compose};
25use super::error::{BundleError, BundleResult};
26use crate::services_root::{ActiveServicesRoot, ServicesProvenance, ServicesRootBootstrap};
27
28use baked::{is_base, stage_baked_base};
29use fetch::{ResolvedSource, SourceContext, resolve_source};
30
31const HTTP_TIMEOUT: Duration = Duration::from_secs(60);
32
33#[derive(Debug, Clone, Copy)]
34pub struct ServicesSourceBootstrap;
35
36impl ServicesSourceBootstrap {
37    pub async fn try_run(
38        profile: &Profile,
39        resolve_secret: impl Fn(&str) -> Option<String> + Send + Sync,
40        core_version: &str,
41    ) -> BundleResult<&'static ActiveServicesRoot> {
42        if let Some(active) = ServicesRootBootstrap::get() {
43            return Ok(active);
44        }
45        Self::resolve(profile, resolve_secret, core_version)
46            .await
47            .map(ServicesRootBootstrap::install)
48    }
49
50    pub async fn resolve(
51        profile: &Profile,
52        resolve_secret: impl Fn(&str) -> Option<String> + Send + Sync,
53        core_version: &str,
54    ) -> BundleResult<ActiveServicesRoot> {
55        if profile.services.sources.is_empty() {
56            return Ok(ActiveServicesRoot {
57                path: PathBuf::from(&profile.paths.services),
58                provenance: ServicesProvenance::Bundled,
59            });
60        }
61
62        let cache = BundleCache::new(cache_root(profile));
63        match Self::compose_sources(profile, &cache, &resolve_secret, core_version).await {
64            Ok(active) => Ok(active),
65            Err(e) => {
66                tracing::error!(error = %e, "Services bundle refresh failed");
67                Self::fall_back(profile, &cache, &e)
68            },
69        }
70    }
71
72    async fn compose_sources(
73        profile: &Profile,
74        cache: &BundleCache,
75        resolve_secret: &(impl Fn(&str) -> Option<String> + Send + Sync),
76        core_version: &str,
77    ) -> BundleResult<ActiveServicesRoot> {
78        let client = reqwest::Client::builder()
79            .redirect(reqwest::redirect::Policy::none())
80            .timeout(HTTP_TIMEOUT)
81            .build()
82            .map_err(|e| BundleError::policy(format!("http client: {e}")))?;
83
84        let previous = cache.read_state();
85        let ctx = SourceContext {
86            cache,
87            state: &previous,
88            core_version,
89            client: &client,
90        };
91        let mut resolved: Vec<ResolvedSource> = Vec::new();
92        for source in &profile.services.sources {
93            let auth = source.auth_secret().and_then(resolve_secret);
94            resolved.push(resolve_source(source, &ctx, auth).await?);
95        }
96        // Why: a profile that pins only kits names no base; the tree baked
97        // into the image is the base, and it must be member zero so ownership
98        // and the authz reconcile treat every pinned source as a kit.
99        if !resolved
100            .first()
101            .is_some_and(|r| is_base(&r.signed.manifest))
102        {
103            let base = stage_baked_base(
104                cache,
105                std::path::Path::new(&profile.paths.services),
106                core_version,
107            )?;
108            resolved.insert(0, base);
109        }
110
111        let members: Vec<BundleMember<'_>> = resolved
112            .iter()
113            .map(|r| BundleMember {
114                name: r.name.clone(),
115                content_hash: r.content_hash.clone(),
116                manifest: &r.signed.manifest,
117            })
118            .collect();
119        let (composed_path, composed_hash) = compose(cache, &members)?;
120        cache.swap_current(&composed_path)?;
121
122        let state = ServicesBundleState {
123            composed_hash: composed_hash.clone(),
124            last_reconciled_hash: previous.last_reconciled_hash.clone(),
125            sources: resolved
126                .iter()
127                .map(|r| (r.name.clone(), r.state.clone()))
128                .collect(),
129        };
130        cache.write_state(&state)?;
131        cache.gc(&composed_hash)?;
132
133        let versions: BTreeMap<String, String> = resolved
134            .iter()
135            .map(|r| (r.name.clone(), r.signed.manifest.version.clone()))
136            .collect();
137        tracing::info!(composed_hash = %composed_hash, sources = resolved.len(), "Services bundles composed");
138
139        Ok(ActiveServicesRoot {
140            path: cache.current_link(),
141            provenance: ServicesProvenance::Fetched {
142                composed_hash,
143                versions,
144            },
145        })
146    }
147
148    fn fall_back(
149        profile: &Profile,
150        cache: &BundleCache,
151        error: &BundleError,
152    ) -> BundleResult<ActiveServicesRoot> {
153        match profile.services.on_fetch_failure {
154            FetchFailurePolicy::FailClosed => Err(BundleError::policy(format!(
155                "services.on_fetch_failure is fail_closed: {error}"
156            ))),
157            FetchFailurePolicy::UseLastGood => {
158                last_good(cache, error).map_or_else(|| bundled_fallback(profile, error), Ok)
159            },
160            FetchFailurePolicy::UseBundled => bundled_fallback(profile, error),
161        }
162    }
163}
164
165fn last_good(cache: &BundleCache, error: &BundleError) -> Option<ActiveServicesRoot> {
166    let current = cache.current_root()?;
167    let state = cache.read_state();
168    if state.composed_hash.is_empty() {
169        return None;
170    }
171    tracing::error!(
172        composed_hash = %state.composed_hash,
173        error = %error,
174        "Serving the last-good services composition after a failed refresh"
175    );
176    Some(ActiveServicesRoot {
177        path: current,
178        provenance: ServicesProvenance::LastGood {
179            composed_hash: state.composed_hash,
180            error: error.to_string(),
181        },
182    })
183}
184
185fn bundled_fallback(profile: &Profile, error: &BundleError) -> BundleResult<ActiveServicesRoot> {
186    let root = PathBuf::from(&profile.paths.services);
187    if !root.join("config/config.yaml").is_file() {
188        return Err(BundleError::policy(format!(
189            "no cached bundle and no baked services tree at {}: {error}",
190            root.display()
191        )));
192    }
193    tracing::error!(
194        path = %root.display(),
195        error = %error,
196        "Serving the services tree baked into the image after a failed refresh"
197    );
198    Ok(ActiveServicesRoot {
199        path: root,
200        provenance: ServicesProvenance::BundledFallback {
201            error: error.to_string(),
202        },
203    })
204}
205
206#[must_use]
207pub fn cache_root(profile: &Profile) -> PathBuf {
208    profile.services.cache_dir.as_ref().map_or_else(
209        || PathBuf::from(&profile.paths.system).join("services-cache"),
210        PathBuf::from,
211    )
212}
systemprompt_loader/bundle/bootstrap/mod.rs

systemprompt_loader/bundle/bootstrap/
mod.rs
