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