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

systemprompt_loader/bundle/bootstrap/
mod.rs
