systemprompt_loader/bundle/bootstrap/
mod.rs1pub 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 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}