bamboo_server/plugin_installer.rs
1//! `ServerPluginInstaller` — the `AppState`-backed implementation of
2//! `bamboo_plugin::PluginInstaller` (Wave 2 § Installer-core agent,
3//! `PLUGIN_PLAN.md`).
4//!
5//! `bamboo-plugin` is an `infra`-layer crate with no access to `AppState`, so
6//! its `LocalPluginInstaller` reference skeleton stops at
7//! `PluginError::NotImplemented` exactly where capability registration needs
8//! `config.json`, `mcp_manager`, `prompt-presets.json`, and
9//! `workflows_dir()`. This type is the real implementation: an ordinary
10//! downstream `impl PluginInstaller for ServerPluginInstaller` (the trait is
11//! foreign, the type is local — no orphan-rule issue).
12//!
13//! # Why a borrowed `web::Data<AppState>` and no `AppState` struct change
14//!
15//! `ServerPluginInstaller` holds a `web::Data<AppState>` clone — the exact
16//! handle every HTTP handler in this crate already receives as an argument
17//! (`web::Data` is `Arc`-backed, so cloning it is cheap). An HTTP handler
18//! constructs one per request: `ServerPluginInstaller::new(state.clone())`.
19//! `AppState` itself is intentionally untouched — no new field, no
20//! coordinated append to `app_state/mod.rs` / `app_state/builder.rs` — so
21//! this branch can never conflict with the other Wave-2 branches that also
22//! stack on `feat/plugin-framework`.
23//!
24//! # Path derivation: `state.app_data_dir`, not the `bamboo_config::paths` globals
25//!
26//! `bamboo_config::paths::{plugins_dir, workflows_dir, plugins_installed_json_path, ...}`
27//! all resolve through a process-wide `OnceLock` that `AppState::new` seeds
28//! ONCE per process (first caller wins — see its doc comment). That is
29//! correct for the single production `AppState` per process, but this
30//! crate's own test suite already builds many `AppState`s over different
31//! `tempfile::tempdir()`s in the same test binary (e.g.
32//! `app_state::tests::test_app_state_creation` and friends) — if this type
33//! read the global helpers, every one of those `AppState`s would silently
34//! share whichever tempdir happened to construct the first one. Every path
35//! below is instead derived from the borrowed `state.app_data_dir` field
36//! directly, exactly the pattern `handlers::settings::workflows` and
37//! `prompt_presets::storage::store_file_path` already use. In production,
38//! where there is exactly one `AppState`, this resolves to the identical
39//! path the global helpers would have produced.
40//!
41//! # Concurrency
42//!
43//! Every `install`/`uninstall` runs under a single process-wide async lock
44//! ([`PLUGIN_OP_LOCK`]), held for the ENTIRE operation including rollback, so
45//! the reconcile→mutate→provenance sequence is atomic w.r.t. any other plugin
46//! op. This closes three concurrency gaps at once: the `installed.json` and
47//! `prompt-presets.json` load/modify/save lost-update races, and the MCP
48//! reconcile→config-write TOCTOU. As additional defense against a concurrent
49//! NON-plugin config write (which does not take this lock), the MCP step also
50//! RE-runs its ownership pre-check INSIDE the `update_config` closure, under
51//! `config_io_lock`, and aborts rather than clobbering if a foreign entry
52//! appeared. Lock ordering is `PLUGIN_OP_LOCK` → `config_io_lock` (never the
53//! reverse) — see [`PLUGIN_OP_LOCK`].
54//!
55//! Narrower, accepted gap: [`ServerPluginInstaller::register_workflows`] does
56//! NOT get the equivalent live re-check that MCP's step does — there is no
57//! `workflows_io_lock` analogous to `config_io_lock` to re-run
58//! `reconcile_exclusive` inside, because workflow files are plain files under
59//! `workflows_dir()`, not a single locked/rewritten document like
60//! `config.json`. So a concurrent NON-plugin write of a same-named workflow
61//! file landing in the window between `existing_workflow_filenames()` and the
62//! actual `fs::write` below could still be clobbered (and then recorded as
63//! plugin-owned, which a later uninstall would incorrectly delete). Deferred:
64//! this is the same class of race the MCP re-check closes, just for a store
65//! that has no equivalent lock to hook into yet.
66//!
67//! # Crash safety (process killed mid-install)
68//!
69//! In-process rollback (below) only fires on an `Err`. A HARD kill after the
70//! MCP step wrote to `config.json` but before provenance is committed would,
71//! without a journal, leave: `reconcile_exclusive` seeing the orphaned mcp id
72//! as existing-but-not-owned → a false `Conflict` on the retry, AND
73//! `uninstall` returning `NotFound` (no provenance) → the user stuck
74//! hand-editing `config.json`. To prevent that, `install` writes a provenance
75//! row with status [`PluginInstallStatus::Installing`] — recording the
76//! INTENDED ownership set — BEFORE steps 1-4, and flips it to
77//! [`PluginInstallStatus::Installed`] only after step 5 succeeds. On the next
78//! install/upgrade of an id whose row is still `Installing` (a prior crash),
79//! [`load_previous_for_disposition`] returns it as `previous` (it does NOT
80//! trip `AlreadyInstalled`), so its intended set is treated as
81//! this-plugin-owned — the leftover reads as an `OwnedReinstall`, not a
82//! foreign conflict — and is cleaned up as an upgrade-over-incomplete.
83//! `uninstall` works on an `Installing` row too.
84//!
85//! # Atomicity / rollback semantics
86//!
87//! `install()` follows `PLUGIN_PLAN.md`'s numbered sequence exactly:
88//!
89//! 0. **Upgrade drop-diff** (only when upgrading an already-installed id):
90//! de-register whatever the new manifest no longer declares, computed via
91//! [`bamboo_plugin::registry::RegisteredCapabilities::removed_since`],
92//! BEFORE registering anything new. De-registration is idempotent/
93//! best-effort (see [`ServerPluginInstaller::deregister_capabilities`]) —
94//! an entry a user already removed by hand never blocks an upgrade.
95//! 1. **MCP** — ownership-checked (REFUSE on a foreign conflict, via
96//! [`bamboo_plugin::registry::reconcile_exclusive`]), merged into
97//! `config.json`, started.
98//! 2. **Prompts** — rename-on-collision (never refuse), appended to
99//! `prompt-presets.json`.
100//! 3. **Workflows** — ownership-checked exactly like MCP, copied into
101//! `workflows_dir()`.
102//! 4. **Skills** — nothing to register (discovered in place); just recorded.
103//! 5. **Provenance commit** — `installed.json` is only ever upserted after
104//! steps 0-4 all succeed.
105//!
106//! Steps 1-3 are real, sequential mutations (config write, then file
107//! writes) — NOT a dry-run computed up front — because `PLUGIN_PLAN.md`
108//! requires the ownership pre-checks to run in that exact order against the
109//! LIVE state each step leaves behind. That means a HARD failure at step 2
110//! or 3 (e.g. an MCP conflict is already past, but the workflow conflict
111//! check at step 3 fails) can happen after step 1 already wrote real
112//! entries into `config.json`. [`ServerPluginInstaller::install`] tracks
113//! every already-applied mutation in an [`InstallRollback`] and, on any hard
114//! failure from steps 1-3, best-effort UNDOES them (removes the mcp entries
115//! it just added and stops any it started, removes the presets it just
116//! appended, deletes the workflow files it just copied) before returning the
117//! error — so a caller's retry starts from a clean slate. Provenance is
118//! never written on a failed path (step 5 is the only place `installed.json`
119//! is touched on success), which is the minimum safety bar even if a rollback
120//! step itself only partially succeeds (rollback operations are themselves
121//! idempotent/log-and-continue, so a second rollback attempt via a plain
122//! retry can never fail louder than the first).
123//!
124//! One known, accepted gap: `stage_plugin_source`/`install_plugin_from_source`
125//! in [`crate::plugin_source`] additionally guard the ON-DISK `plugin_dir`
126//! swap itself (an upgrade's new bundle replaces the old one's files at a
127//! fixed path) by moving the previous bundle aside instead of deleting it, and
128//! restoring it if `install()` subsequently fails — see that module's docs.
129//!
130//! Prompt-preset drop-diff caveat: the upgrade drop-diff compares the NEW
131//! manifest's nominal preset ids against the OLD install's ACTUAL (possibly
132//! renamed-on-collision) registered ids. A preset that got renamed at its
133//! original install time and is still declared under its original nominal id
134//! in the new manifest will look "dropped" (the nominal id is absent from the
135//! actual old set) and get re-appended (possibly renamed again). This is
136//! harmless — preset content is just refreshed under a fresh id — and not
137//! worth a stable-id-mapping schema change for what `RegisteredCapabilities`
138//! already documents as the one rename (not refuse) exception.
139
140use std::collections::HashSet;
141use std::path::{Path, PathBuf};
142
143use async_trait::async_trait;
144use chrono::{DateTime, Utc};
145use tokio::fs;
146
147use bamboo_domain::mcp_config::McpServerConfig;
148use bamboo_plugin::installer::{load_previous_for_disposition, preflight_install};
149use bamboo_plugin::manifest::{Platform, ServiceManifestEntry};
150use bamboo_plugin::registry::{reconcile_exclusive, RegisteredCapabilities};
151use bamboo_plugin::{
152 InstallDisposition, InstalledPlugin, InstalledPlugins, PluginError, PluginInstallStatus,
153 PluginInstaller, PluginManifest, PluginResult, PluginSource,
154};
155
156use crate::app_state::{AppState, ConfigUpdateEffects};
157use crate::error::AppError;
158use crate::handlers::agent::mcp::upsert_server_by_id;
159use crate::handlers::agent::prompt_presets::{
160 ensure_unique_preset_id, load_store, save_store, store_file_path, StoredPromptPreset,
161};
162use crate::service_manager::{ServiceManager, ServiceRuntimeConfig};
163
164/// Process-wide serialization of plugin install/uninstall operations.
165///
166/// The whole ownership/upgrade machinery is a read-modify-write over shared
167/// stores (`config.json`, `prompt-presets.json`, `installed.json`) with the
168/// ownership pre-check and the eventual mutation in separate steps. Under
169/// CONCURRENT plugin ops (the HTTP agent will expose exactly that) those
170/// interleave badly: two installs of different ids race `installed.json`'s
171/// load/add/save (last save drops the other's row), `prompt-presets.json`'s
172/// load/save (lost update), and the MCP reconcile→write window (a foreign
173/// entry landing mid-window gets clobbered AND recorded as plugin-owned,
174/// re-opening BLOCKER-1). Plugin installs are rare and not perf-sensitive, so
175/// one coarse process-wide lock held across the ENTIRE `install`/`uninstall`
176/// (including rollback) is the right call — it makes each op's
177/// reconcile→mutate→provenance sequence atomic w.r.t. every other plugin op.
178///
179/// Lock ordering: this lock is acquired at the TOP of `install`/`uninstall`,
180/// OUTSIDE any `AppState::update_config` call (which internally takes
181/// `config_io_lock`). So the order is always `PLUGIN_OP_LOCK` →
182/// `config_io_lock`, never the reverse — no deadlock. Nothing acquires
183/// `PLUGIN_OP_LOCK` while holding `config_io_lock`.
184///
185/// # Single-process assumption (deferred: no cross-process lock)
186///
187/// This is a `tokio::sync::Mutex` — IN-PROCESS only. It serializes plugin ops
188/// within one `bamboo serve` process, but two SEPARATE `bamboo serve`
189/// processes pointed at the same `~/.bamboo` data dir would each get their
190/// own independent `PLUGIN_OP_LOCK` and could race each other's
191/// reconcile→mutate→provenance sequence exactly the way this lock exists to
192/// prevent for concurrent ops WITHIN one process. The plugin system assumes
193/// the normal deployment: a SINGLE `bamboo serve` per data directory. True
194/// multi-process safety would need an OS-level file lock (e.g. `flock` on a
195/// lockfile under `plugins_dir()`) instead of/in addition to this `Mutex`;
196/// that's a documented follow-up, not implemented here.
197static PLUGIN_OP_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
198
199/// AppState-backed [`PluginInstaller`]. See the module docs for the full
200/// design rationale (borrowing, path derivation, atomicity).
201pub struct ServerPluginInstaller {
202 state: actix_web::web::Data<AppState>,
203}
204
205/// Mutations already applied by a not-yet-committed `install()`, so a hard
206/// failure partway through steps 1-3 can best-effort undo exactly what has
207/// been done so far. See the module docs' "Atomicity / rollback semantics".
208#[derive(Default)]
209struct InstallRollback {
210 mcp_ids_added: Vec<String>,
211 mcp_ids_started: Vec<String>,
212 preset_ids_added: Vec<String>,
213 workflow_files_added: Vec<String>,
214 /// Service ids this install claimed ownership of (whether or not the
215 /// actual `start_service` call succeeded — best-effort, same contract as
216 /// `mcp_ids_added`/`mcp_ids_started`).
217 service_ids_added: Vec<String>,
218 /// Subset of `service_ids_added` that actually got a running
219 /// `ServiceManager` runtime started — only these need `stop_service` on
220 /// rollback.
221 service_ids_started: Vec<String>,
222}
223
224impl ServerPluginInstaller {
225 pub fn new(state: actix_web::web::Data<AppState>) -> Self {
226 Self { state }
227 }
228
229 fn plugins_dir(&self) -> PathBuf {
230 self.state.app_data_dir.join("plugins")
231 }
232
233 fn installed_json_path(&self) -> PathBuf {
234 self.plugins_dir().join("installed.json")
235 }
236
237 fn workflows_dir(&self) -> PathBuf {
238 self.state.app_data_dir.join("workflows")
239 }
240
241 fn prompt_presets_path(&self) -> PathBuf {
242 store_file_path(&self.state.app_data_dir)
243 }
244
245 /// `<data_dir>/plugin_service_config/<plugin_id>/config.json` — the
246 /// per-service user config path passed to services as
247 /// `BAMBOO_PLUGIN_SERVICE_CONFIG` (issue #479 open question 2).
248 ///
249 /// Deliberately NOT under `plugins_dir()/<plugin_id>/` (the
250 /// swap-managed `plugin_dir`): `plugin_source::stage_plugin_source`
251 /// upgrades a plugin by renaming the ENTIRE old `plugin_dir` aside and
252 /// swapping a freshly-staged directory into its place — any file living
253 /// inside `plugin_dir` would be swept away with the old bundle on
254 /// upgrade (or deleted outright on uninstall) unless bamboo specifically
255 /// carried it forward, which it does not. A sibling directory, named
256 /// only by `plugin_id`, is untouched by that swap and by
257 /// `uninstall`'s `remove_dir_all(plugin_dir)` — so a service's own
258 /// config (which may carry tokens/secrets a connector needs) survives
259 /// both an upgrade and an uninstall. bamboo only ever creates the PARENT
260 /// directory here (see [`Self::ensure_service_config_parent_dir`]) —
261 /// never writes or deletes `config.json` itself; on uninstall it is
262 /// deliberately left in place (not part of `remove_dir_all`), so a
263 /// later re-install of the same plugin id picks its old config back up
264 /// automatically.
265 fn service_config_path(&self, plugin_id: &str) -> PathBuf {
266 service_config_path_under(&self.state.app_data_dir, plugin_id)
267 }
268
269 async fn ensure_service_config_parent_dir(&self, plugin_id: &str) -> PluginResult<()> {
270 let path = self.service_config_path(plugin_id);
271 if let Some(parent) = path.parent() {
272 fs::create_dir_all(parent).await?;
273 }
274 Ok(())
275 }
276
277 /// Every service id currently owned by any OTHER installed plugin — the
278 /// "existing" side of [`reconcile_exclusive`] for services. There is no
279 /// single shared document analogous to `config.json` for services, so
280 /// provenance itself is the source of truth for "who owns this id".
281 ///
282 /// `exclude_plugin_id` MUST be the id of the plugin currently being
283 /// installed/upgraded and is always excluded — NOT an optional
284 /// nicety: by the time [`Self::register_services`] calls this,
285 /// `install()`'s crash-safety journal write has ALREADY upserted THIS
286 /// plugin's `Installing` row into `installed.json` with its full
287 /// INTENDED `service_ids` (see `install()`'s "Crash-safety journal"
288 /// step, which runs before every registration step). Without excluding
289 /// it, a plain fresh install would see its own not-yet-committed row as
290 /// a foreign owner of its own declared ids and refuse itself.
291 /// `previously_owned` (computed separately, from `previous` BEFORE that
292 /// journal write) is what still correctly classifies an upgrade's
293 /// re-declared ids as `OwnedReinstall` rather than `New`.
294 async fn existing_service_ids(&self, exclude_plugin_id: &str) -> PluginResult<Vec<String>> {
295 let store = InstalledPlugins::load(&self.installed_json_path()).await?;
296 Ok(store
297 .list()
298 .iter()
299 .filter(|plugin| plugin.id != exclude_plugin_id)
300 .flat_map(|plugin| plugin.registered.service_ids.iter().cloned())
301 .collect())
302 }
303
304 /// Resolve one manifest-declared service entry into a
305 /// [`ServiceRuntimeConfig`] ready for `ServiceManager::start_service`.
306 fn resolve_service_config(
307 &self,
308 plugin_id: &str,
309 entry: &ServiceManifestEntry,
310 plugin_dir: &Path,
311 platform: Platform,
312 ) -> ServiceRuntimeConfig {
313 resolve_service_config_under(
314 &self.state.app_data_dir,
315 plugin_id,
316 entry,
317 plugin_dir,
318 platform,
319 )
320 }
321
322 /// Every `.md` filename directly under `workflows_dir()` (created if
323 /// missing). Mirrors `handlers::settings::workflows::list_workflows`'s
324 /// listing logic but returns bare filenames for
325 /// [`bamboo_plugin::registry::reconcile_exclusive`].
326 async fn existing_workflow_filenames(&self) -> PluginResult<Vec<String>> {
327 let dir = self.workflows_dir();
328 fs::create_dir_all(&dir).await?;
329 let mut entries = fs::read_dir(&dir).await?;
330 let mut names = Vec::new();
331 while let Some(entry) = entries.next_entry().await? {
332 let is_file = entry
333 .file_type()
334 .await
335 .map(|file_type| file_type.is_file())
336 .unwrap_or(false);
337 if !is_file {
338 continue;
339 }
340 if let Some(name) = entry.file_name().to_str() {
341 if name.ends_with(".md") {
342 names.push(name.to_string());
343 }
344 }
345 }
346 Ok(names)
347 }
348
349 // --- De-registration primitives (shared by upgrade drop-diff, install
350 // rollback, and `uninstall`). Each is individually idempotent/tolerant —
351 // an entry that is already gone (e.g. a user manually deleted it) is
352 // logged and skipped, never a hard failure — matching the requirement
353 // that de-registration never blocks an uninstall/upgrade retry. ---
354
355 async fn remove_mcp_server(&self, id: &str) {
356 let owned_id = id.to_string();
357 let result = self
358 .state
359 .update_config(
360 move |cfg| {
361 cfg.mcp.servers.retain(|server| server.id != owned_id);
362 Ok(())
363 },
364 ConfigUpdateEffects::default(),
365 )
366 .await;
367 if let Err(error) = result {
368 tracing::warn!(
369 mcp_server_id = %id,
370 %error,
371 "failed to remove plugin-owned mcp server from config.json; continuing"
372 );
373 }
374 if let Err(error) = self.state.mcp_manager.stop_server(id).await {
375 tracing::warn!(
376 mcp_server_id = %id,
377 %error,
378 "failed to stop plugin-owned mcp server; continuing"
379 );
380 }
381 }
382
383 async fn remove_prompt_preset(&self, preset_id: &str) {
384 let path = self.prompt_presets_path();
385 match load_store(&path).await {
386 Ok(mut store) => {
387 let before = store.prompts.len();
388 store.prompts.retain(|preset| preset.id != preset_id);
389 if store.prompts.len() != before {
390 if let Err(error) = save_store(&path, &store).await {
391 tracing::warn!(
392 %preset_id,
393 %error,
394 "failed to persist prompt-presets.json after removing plugin-owned preset; continuing"
395 );
396 }
397 }
398 }
399 Err(error) => {
400 tracing::warn!(
401 %preset_id,
402 %error,
403 "failed to load prompt-presets.json while removing plugin-owned preset; continuing"
404 );
405 }
406 }
407 }
408
409 async fn remove_service(&self, id: &str) {
410 if let Err(error) = self.state.service_manager.stop_service(id).await {
411 tracing::warn!(
412 service_id = %id,
413 %error,
414 "failed to stop plugin-owned service; continuing"
415 );
416 }
417 }
418
419 async fn remove_workflow_file(&self, filename: &str) {
420 let path = self.workflows_dir().join(filename);
421 match fs::remove_file(&path).await {
422 Ok(()) => {}
423 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
424 Err(error) => {
425 tracing::warn!(
426 %filename,
427 %error,
428 "failed to remove plugin-owned workflow file; continuing"
429 );
430 }
431 }
432 }
433
434 /// De-register a whole [`RegisteredCapabilities`] set (used for the
435 /// upgrade drop-diff and for `uninstall`). Skill dirs need no shared-store
436 /// action — they are only ever removed by deleting `plugin_dir` itself.
437 async fn deregister_capabilities(&self, registered: &RegisteredCapabilities) {
438 for mcp_id in ®istered.mcp_server_ids {
439 self.remove_mcp_server(mcp_id).await;
440 }
441 for preset_id in ®istered.preset_ids {
442 self.remove_prompt_preset(preset_id).await;
443 }
444 for workflow_filename in ®istered.workflow_filenames {
445 self.remove_workflow_file(workflow_filename).await;
446 }
447 for service_id in ®istered.service_ids {
448 self.remove_service(service_id).await;
449 }
450 }
451
452 /// Best-effort undo of an `install()` that failed partway through steps
453 /// 1-3. See the module docs.
454 async fn rollback_partial_install(&self, rollback: &InstallRollback) {
455 for id in &rollback.mcp_ids_started {
456 let _ = self.state.mcp_manager.stop_server(id).await;
457 }
458 for id in &rollback.mcp_ids_added {
459 self.remove_mcp_server(id).await;
460 }
461 for id in &rollback.preset_ids_added {
462 self.remove_prompt_preset(id).await;
463 }
464 for file in &rollback.workflow_files_added {
465 self.remove_workflow_file(file).await;
466 }
467 for id in &rollback.service_ids_started {
468 let _ = self.state.service_manager.stop_service(id).await;
469 }
470 }
471
472 /// Upsert one provenance row into `installed.json`. Used both for the
473 /// pre-registration `Installing` journal row and the final `Installed`
474 /// commit — the ONLY two writers of `installed.json` in `install`. Both
475 /// run under [`PLUGIN_OP_LOCK`], so the load/add/save is race-free.
476 async fn upsert_provenance(&self, entry: InstalledPlugin, path: &Path) -> PluginResult<()> {
477 let mut store = InstalledPlugins::load(path).await?;
478 store.add(entry);
479 store.save(path).await?;
480 Ok(())
481 }
482
483 /// Abort an in-process `install` failure: best-effort undo of the partial
484 /// registration (steps 1-3), then restore `installed.json` to its
485 /// pre-install state — re-writing the original `previous` row on an
486 /// upgrade/recovery, or removing the id's row entirely on a fresh install
487 /// — so the `Installing` journal row we wrote up front never lingers after
488 /// a clean in-process failure. (A HARD kill is the only path that
489 /// intentionally leaves an `Installing` row, for the next op to recover.)
490 async fn abort_install(
491 &self,
492 rollback: &InstallRollback,
493 previous: &Option<InstalledPlugin>,
494 plugin_id: &str,
495 path: &Path,
496 ) {
497 self.rollback_partial_install(rollback).await;
498 let restore = match previous {
499 Some(prev) => self.upsert_provenance(prev.clone(), path).await,
500 None => match InstalledPlugins::load(path).await {
501 Ok(mut store) => {
502 store.remove(plugin_id);
503 store.save(path).await
504 }
505 Err(error) => Err(error),
506 },
507 };
508 if let Err(error) = restore {
509 tracing::warn!(
510 %plugin_id,
511 %error,
512 "failed to restore provenance after aborting a failed install; a stale \
513 `installing` row may remain (recoverable by a retry)"
514 );
515 }
516 }
517
518 /// Step 1: MCP. Returns the ids actually (re-)registered
519 /// (`reconciliation.to_register`), which — once past the conflict gate —
520 /// is exactly the declared id set.
521 async fn register_mcp(
522 &self,
523 manifest: &PluginManifest,
524 resolved_mcp_servers: Vec<McpServerConfig>,
525 previously_owned: &[String],
526 rollback: &mut InstallRollback,
527 ) -> PluginResult<Vec<String>> {
528 if resolved_mcp_servers.is_empty() {
529 return Ok(Vec::new());
530 }
531
532 let declared_ids: Vec<String> = manifest
533 .provides
534 .mcp_servers
535 .iter()
536 .map(|entry| entry.id.clone())
537 .collect();
538 let existing_ids: Vec<String> = {
539 let config = self.state.config.read().await;
540 config.mcp.servers.iter().map(|s| s.id.clone()).collect()
541 };
542
543 let reconciliation = reconcile_exclusive(&declared_ids, &existing_ids, previously_owned);
544 if !reconciliation.foreign_conflicts.is_empty() {
545 return Err(PluginError::Conflict {
546 kind: "mcp server",
547 name: reconciliation.foreign_conflicts.join(", "),
548 plugin_id: manifest.id.clone(),
549 });
550 }
551
552 let to_register: HashSet<&str> = reconciliation
553 .to_register
554 .iter()
555 .map(String::as_str)
556 .collect();
557 let configs_to_register: Vec<McpServerConfig> = resolved_mcp_servers
558 .into_iter()
559 .filter(|config| to_register.contains(config.id.as_str()))
560 .collect();
561
562 let owned_configs = configs_to_register.clone();
563 let declared_for_recheck = declared_ids.clone();
564 let owned_for_recheck: Vec<String> = previously_owned.to_vec();
565 let plugin_id_for_recheck = manifest.id.clone();
566 self.state
567 .update_config(
568 move |cfg| {
569 // TOCTOU guard: re-run the ownership pre-check against the
570 // LIVE config while holding config_io_lock, so a foreign
571 // entry that landed between our earlier read and now can't
572 // be silently clobbered (and then recorded as
573 // plugin-owned, re-opening BLOCKER-1 under a race).
574 // Concurrent PLUGIN ops are already excluded by
575 // PLUGIN_OP_LOCK; this closes the residual window against a
576 // concurrent NON-plugin config write.
577 let live_existing: Vec<String> =
578 cfg.mcp.servers.iter().map(|s| s.id.clone()).collect();
579 let live = reconcile_exclusive(
580 &declared_for_recheck,
581 &live_existing,
582 &owned_for_recheck,
583 );
584 if !live.foreign_conflicts.is_empty() {
585 return Err(AppError::BadRequest(format!(
586 "mcp server(s) '{}' now conflict with a non-plugin entry (a concurrent \
587 change landed mid-install); refusing to overwrite for plugin '{}'",
588 live.foreign_conflicts.join(", "),
589 plugin_id_for_recheck
590 )));
591 }
592 // Shared by-id merge (same helper import_servers uses).
593 for server in &owned_configs {
594 upsert_server_by_id(&mut cfg.mcp.servers, server.clone());
595 }
596 Ok(())
597 },
598 ConfigUpdateEffects::default(),
599 )
600 .await
601 .map_err(|error| {
602 PluginError::Registration(format!("failed to write mcp servers to config: {error}"))
603 })?;
604 // Config write for the whole batch succeeded — record ownership now,
605 // regardless of whether individual `start_server` calls below
606 // succeed (matches `import_servers`' best-effort start semantics: a
607 // config entry that fails to start is still a real, plugin-owned
608 // registration a user/CLI can retry starting later).
609 rollback.mcp_ids_added = reconciliation.to_register.clone();
610
611 for server in &configs_to_register {
612 // Stop any stale running instance first, matching the
613 // update/import handlers' pattern.
614 let _ = self.state.mcp_manager.stop_server(&server.id).await;
615 if server.enabled {
616 match self.state.mcp_manager.start_server(server.clone()).await {
617 Ok(()) => rollback.mcp_ids_started.push(server.id.clone()),
618 Err(error) => tracing::warn!(
619 mcp_server_id = %server.id,
620 %error,
621 "plugin-registered mcp server failed to start; config entry kept (best-effort)"
622 ),
623 }
624 }
625 }
626
627 Ok(reconciliation.to_register)
628 }
629
630 /// Step 1b: Services (issue #479, prereq for epic #477). Same
631 /// REFUSE-on-conflict + best-effort-start shape as [`Self::register_mcp`],
632 /// against [`Self::existing_service_ids`] instead of `config.json` (there
633 /// is no shared config document for services — provenance itself is the
634 /// ownership store, see that method's doc comment).
635 async fn register_services(
636 &self,
637 manifest: &PluginManifest,
638 plugin_dir: &Path,
639 previously_owned: &[String],
640 rollback: &mut InstallRollback,
641 ) -> PluginResult<Vec<String>> {
642 if manifest.provides.services.is_empty() {
643 return Ok(Vec::new());
644 }
645
646 let declared_ids: Vec<String> = manifest
647 .provides
648 .services
649 .iter()
650 .map(|entry| entry.id.clone())
651 .collect();
652 let existing_ids = self.existing_service_ids(&manifest.id).await?;
653 let reconciliation = reconcile_exclusive(&declared_ids, &existing_ids, previously_owned);
654 if !reconciliation.foreign_conflicts.is_empty() {
655 return Err(PluginError::Conflict {
656 kind: "service",
657 name: reconciliation.foreign_conflicts.join(", "),
658 plugin_id: manifest.id.clone(),
659 });
660 }
661
662 // Ownership is claimed regardless of individual start outcomes below
663 // (matches `register_mcp`'s "config write succeeded, so record
664 // ownership" contract — here there is no config write, so this is
665 // simply claimed up front).
666 rollback.service_ids_added = reconciliation.to_register.clone();
667
668 self.ensure_service_config_parent_dir(&manifest.id).await?;
669 let platform = Platform::current().unwrap_or(Platform::Linux);
670 let to_register: HashSet<&str> = reconciliation
671 .to_register
672 .iter()
673 .map(String::as_str)
674 .collect();
675
676 for entry in &manifest.provides.services {
677 if !to_register.contains(entry.id.as_str()) {
678 continue;
679 }
680 // Stop any stale running instance first — covers a leftover
681 // from a crashed install/upgrade recovery (a genuine same-id
682 // upgrade already had its old service stopped BEFORE staging by
683 // `stop_services_for_upgrade`, see the module docs' "Same-id
684 // upgrade ordering").
685 let _ = self.state.service_manager.stop_service(&entry.id).await;
686 if !entry.enabled {
687 continue;
688 }
689 let config = self.resolve_service_config(&manifest.id, entry, plugin_dir, platform);
690 match self.state.service_manager.start_service(config).await {
691 Ok(()) => rollback.service_ids_started.push(entry.id.clone()),
692 Err(error) => tracing::warn!(
693 service_id = %entry.id,
694 %error,
695 "plugin-registered service failed to start; ownership kept (best-effort, matches mcp)"
696 ),
697 }
698 }
699
700 Ok(reconciliation.to_register)
701 }
702
703 /// Step 2: Prompts. Rename-on-collision (never refuses) — returns the
704 /// ACTUAL ids used (after any rename), which is what provenance must
705 /// record.
706 async fn register_prompts(&self, manifest: &PluginManifest) -> PluginResult<Vec<String>> {
707 if manifest.provides.prompts.is_empty() {
708 return Ok(Vec::new());
709 }
710
711 let path = self.prompt_presets_path();
712 let mut store = load_store(&path).await.map_err(|error| {
713 PluginError::Registration(format!("failed to load prompt-presets.json: {error}"))
714 })?;
715
716 let mut existing_ids: HashSet<String> = store
717 .prompts
718 .iter()
719 .map(|preset| preset.id.clone())
720 .collect();
721 // `general_assistant` (bamboo-server's DEFAULT_PRESET_ID) is never a
722 // row in the store, so it wouldn't otherwise appear in `existing_ids`
723 // — but manifest validation already rejects any plugin declaring it
724 // (RESERVED_PRESET_IDS), so no extra guard is needed here.
725
726 let mut actual_ids = Vec::with_capacity(manifest.provides.prompts.len());
727 for preset in &manifest.provides.prompts {
728 let actual_id = ensure_unique_preset_id(&preset.id, &existing_ids);
729 store.prompts.push(StoredPromptPreset {
730 id: actual_id.clone(),
731 name: preset.name.clone(),
732 description: preset.description.clone(),
733 content: preset.content.clone(),
734 });
735 existing_ids.insert(actual_id.clone());
736 actual_ids.push(actual_id);
737 }
738
739 save_store(&path, &store).await.map_err(|error| {
740 PluginError::Registration(format!("failed to persist prompt-presets.json: {error}"))
741 })?;
742
743 Ok(actual_ids)
744 }
745
746 /// Step 3: Workflows. Same REFUSE-on-conflict shape as MCP.
747 ///
748 /// Takes `rollback` directly (unlike [`Self::register_prompts`], which
749 /// commits in one atomic file write) because each workflow file is
750 /// copied with a SEPARATE `fs::write` call: if copying the Nth file
751 /// fails, files 1..N-1 are already really on disk, and `rollback` must
752 /// know about them even though this function returns `Err` — recording
753 /// the whole `to_register` list only on a successful `Ok` return (the
754 /// pattern the caller uses for [`Self::register_mcp`]/
755 /// [`Self::register_prompts`]) would lose that partial progress.
756 async fn register_workflows(
757 &self,
758 manifest: &PluginManifest,
759 plugin_dir: &Path,
760 previously_owned: &[String],
761 rollback: &mut InstallRollback,
762 ) -> PluginResult<Vec<String>> {
763 if manifest.provides.workflows.is_empty() {
764 return Ok(Vec::new());
765 }
766
767 let declared: Vec<String> = manifest.provides.workflows.clone();
768 let existing = self.existing_workflow_filenames().await?;
769 let reconciliation = reconcile_exclusive(&declared, &existing, previously_owned);
770 if !reconciliation.foreign_conflicts.is_empty() {
771 return Err(PluginError::Conflict {
772 kind: "workflow",
773 name: reconciliation.foreign_conflicts.join(", "),
774 plugin_id: manifest.id.clone(),
775 });
776 }
777
778 let dest_dir = self.workflows_dir();
779 for filename in &reconciliation.to_register {
780 let stem = filename.strip_suffix(".md").unwrap_or(filename);
781 if !bamboo_config::paths::is_safe_workflow_name(stem) {
782 return Err(PluginError::InvalidManifest(format!(
783 "workflow filename '{filename}' is not a safe workflow name"
784 )));
785 }
786 let source_path = plugin_dir.join("workflows").join(filename);
787 let content = fs::read_to_string(&source_path).await?;
788 fs::write(dest_dir.join(filename), content).await?;
789 // Recorded immediately, not after the whole loop: if a LATER
790 // file in this same call fails, this one is already really on
791 // disk and rollback must know to remove it.
792 rollback.workflow_files_added.push(filename.clone());
793 }
794
795 Ok(reconciliation.to_register)
796 }
797
798 /// **Same-id upgrade ordering** (issue #479 "Install-flow deltas" /
799 /// "Same-id upgrade ordering bug risk"): `plugin_source::stage_plugin_source`
800 /// swaps `plugin_dir`'s ENTIRE contents (old bundle moved to a
801 /// `.backup-*` dir, staged bundle renamed into place) BEFORE
802 /// `install()` — and therefore [`Self::register_services`] — ever runs.
803 /// A still-running old service process holding the old binary open
804 /// during that swap is at best running stale code post-swap and at
805 /// worst (Windows) blocks the rename outright. The minimal seam that
806 /// fixes the ordering without restructuring `stage_plugin_source`/
807 /// `install()`: the HTTP `update_plugin` handler calls this BEFORE
808 /// `stage_plugin_source`, using the URL path's target id (known up
809 /// front for an upgrade, unlike a fresh `install`) to look up the
810 /// CURRENTLY-installed row's `registered.service_ids` and stop each one.
811 /// Net effect: stop (old binary) → swap (new binary) → start (new
812 /// binary, via `register_services` inside `install()`), exactly the
813 /// sequencing the issue calls for.
814 ///
815 /// Best-effort: returns exactly the ids that were actually running and
816 /// got stopped (not e.g. an already-stopped or unknown id), so a
817 /// subsequently-failed upgrade can restart precisely those — see
818 /// [`Self::restart_services_after_failed_upgrade`]. A plugin with no
819 /// prior install (or no services) returns an empty vec and stops
820 /// nothing.
821 pub async fn stop_services_for_upgrade(&self, plugin_id: &str) -> Vec<String> {
822 let store = match InstalledPlugins::load(&self.installed_json_path()).await {
823 Ok(store) => store,
824 Err(error) => {
825 tracing::warn!(
826 %plugin_id,
827 %error,
828 "stop_services_for_upgrade: failed to load installed.json; skipping"
829 );
830 return Vec::new();
831 }
832 };
833 let Some(entry) = store.get(plugin_id) else {
834 return Vec::new();
835 };
836 let mut stopped = Vec::with_capacity(entry.registered.service_ids.len());
837 for service_id in &entry.registered.service_ids {
838 match self.state.service_manager.stop_service(service_id).await {
839 Ok(()) => stopped.push(service_id.clone()),
840 Err(error) => tracing::debug!(
841 service_id = %service_id,
842 %error,
843 "stop_services_for_upgrade: service was not running; nothing to stop"
844 ),
845 }
846 }
847 stopped
848 }
849
850 /// Counterpart to [`Self::stop_services_for_upgrade`]: called after a
851 /// FAILED upgrade whose `StagedPlugin::rollback()` already restored
852 /// `plugin_dir` to the pre-upgrade bundle's bytes — re-reads that
853 /// (now-restored) OLD `plugin.json` and restarts exactly the services in
854 /// `stopped` that it still declares as `enabled`. Best-effort/
855 /// log-and-continue: a failure here leaves the affected service stopped
856 /// (a degraded-but-safe outcome — never silently double-runs an old and
857 /// a new instance) rather than panicking the request.
858 pub async fn restart_services_after_failed_upgrade(&self, plugin_id: &str, stopped: &[String]) {
859 if stopped.is_empty() {
860 return;
861 }
862 let store = match InstalledPlugins::load(&self.installed_json_path()).await {
863 Ok(store) => store,
864 Err(error) => {
865 tracing::warn!(
866 %plugin_id,
867 %error,
868 "restart_services_after_failed_upgrade: failed to load installed.json"
869 );
870 return;
871 }
872 };
873 let Some(entry) = store.get(plugin_id) else {
874 return;
875 };
876 let manifest_path = entry.plugin_dir.join("plugin.json");
877 let manifest = match fs::read_to_string(&manifest_path)
878 .await
879 .ok()
880 .and_then(|raw| PluginManifest::parse_str(&raw).ok())
881 {
882 Some(manifest) => manifest,
883 None => {
884 tracing::warn!(
885 %plugin_id,
886 path = %manifest_path.display(),
887 "restart_services_after_failed_upgrade: failed to read/parse the \
888 rolled-back plugin.json; affected service(s) remain stopped"
889 );
890 return;
891 }
892 };
893 let platform = Platform::current().unwrap_or(Platform::Linux);
894 for svc in &manifest.provides.services {
895 if !stopped.contains(&svc.id) || !svc.enabled {
896 continue;
897 }
898 let config = self.resolve_service_config(plugin_id, svc, &entry.plugin_dir, platform);
899 if let Err(error) = self.state.service_manager.start_service(config).await {
900 tracing::warn!(
901 service_id = %svc.id,
902 %plugin_id,
903 %error,
904 "failed to restart service after a failed upgrade rolled back to the \
905 previous plugin bundle; service remains stopped"
906 );
907 }
908 }
909 }
910}
911
912#[async_trait]
913impl PluginInstaller for ServerPluginInstaller {
914 async fn install(
915 &self,
916 manifest: &PluginManifest,
917 plugin_dir: &Path,
918 source: PluginSource,
919 disposition: InstallDisposition,
920 installed_at: DateTime<Utc>,
921 ) -> PluginResult<InstalledPlugin> {
922 // Serialize the whole op against every other plugin install/uninstall
923 // (process-wide) — held across all steps AND rollback. See module docs
924 // "Concurrency".
925 let _op_guard = PLUGIN_OP_LOCK.lock().await;
926
927 let installed_json_path = self.installed_json_path();
928
929 // Disposition gate (AlreadyInstalled only for a COMPLETED prior
930 // install; an `Installing` leftover is returned for recovery) + the
931 // rest of the pure, AppState-free validation this crate can already
932 // do (manifest shape, platform gate, on-disk skill/workflow
933 // existence, `provides.skills` authoritativeness).
934 let previous =
935 load_previous_for_disposition(&installed_json_path, &manifest.id, disposition).await?;
936 let resolved_mcp_servers = preflight_install(manifest, plugin_dir).await?;
937
938 // The set this install INTENDS to own, by declaration order. Used both
939 // for the crash-safety journal row (below) and the step-0 drop-diff.
940 let intended = RegisteredCapabilities {
941 mcp_server_ids: manifest
942 .provides
943 .mcp_servers
944 .iter()
945 .map(|entry| entry.id.clone())
946 .collect(),
947 skill_dirs: manifest.provides.skills.clone(),
948 preset_ids: manifest
949 .provides
950 .prompts
951 .iter()
952 .map(|preset| preset.id.clone())
953 .collect(),
954 workflow_filenames: manifest.provides.workflows.clone(),
955 service_ids: manifest
956 .provides
957 .services
958 .iter()
959 .map(|entry| entry.id.clone())
960 .collect(),
961 };
962
963 // Step 0: upgrade drop-diff. Computed from the NEW manifest's plain
964 // declared ids (see module docs re: the preset-rename caveat) vs the
965 // OLD install's registered set — de-register whatever the new version
966 // no longer declares BEFORE registering anything new (BLOCKER 2). Also
967 // fires for a recovery over an `Installing` leftover: its intended set
968 // is diffed the same way, so a crashed attempt's extra ids get cleaned.
969 if let Some(previous) = &previous {
970 let dropped = intended.removed_since(&previous.registered);
971 if !dropped.is_empty() {
972 tracing::info!(
973 plugin_id = %manifest.id,
974 recovering = previous.status == PluginInstallStatus::Installing,
975 dropped_mcp = ?dropped.mcp_server_ids,
976 dropped_presets = ?dropped.preset_ids,
977 dropped_workflows = ?dropped.workflow_filenames,
978 dropped_services = ?dropped.service_ids,
979 "install drop-diff: de-registering capabilities the new/completed version no longer declares"
980 );
981 self.deregister_capabilities(&dropped).await;
982 }
983 }
984
985 let previously_owned_mcp = previous
986 .as_ref()
987 .map(|p| p.registered.mcp_server_ids.clone())
988 .unwrap_or_default();
989 let previously_owned_workflows = previous
990 .as_ref()
991 .map(|p| p.registered.workflow_filenames.clone())
992 .unwrap_or_default();
993 let previously_owned_services = previous
994 .as_ref()
995 .map(|p| p.registered.service_ids.clone())
996 .unwrap_or_default();
997
998 // Crash-safety journal: write an `Installing` provenance row recording
999 // the INTENDED ownership set BEFORE mutating any shared store, so a
1000 // hard kill mid-install leaves a recoverable marker (see module docs
1001 // "Crash safety"). On a fresh install this creates the row; on an
1002 // upgrade/recovery it overwrites the prior row.
1003 self.upsert_provenance(
1004 InstalledPlugin {
1005 id: manifest.id.clone(),
1006 version: manifest.version.clone(),
1007 source: source.clone(),
1008 plugin_dir: plugin_dir.to_path_buf(),
1009 installed_at,
1010 status: PluginInstallStatus::Installing,
1011 registered: intended.clone(),
1012 },
1013 &installed_json_path,
1014 )
1015 .await?;
1016
1017 let mut rollback = InstallRollback::default();
1018
1019 // Step 1: MCP.
1020 let mcp_server_ids = match self
1021 .register_mcp(
1022 manifest,
1023 resolved_mcp_servers,
1024 &previously_owned_mcp,
1025 &mut rollback,
1026 )
1027 .await
1028 {
1029 Ok(ids) => ids,
1030 Err(error) => {
1031 self.abort_install(&rollback, &previous, &manifest.id, &installed_json_path)
1032 .await;
1033 return Err(error);
1034 }
1035 };
1036
1037 // Step 1b: Services (issue #479). Runs right after MCP, before the
1038 // never-refusing Prompts step, so a services conflict fails the
1039 // install as early as the other REFUSE-on-conflict kinds do. Note:
1040 // for a same-id UPGRADE, the OLD service (if any) was already
1041 // stopped by `stop_services_for_upgrade` before `stage_plugin_source`
1042 // swapped `plugin_dir` — see that method's doc comment on the
1043 // stop→swap→start sequencing.
1044 let service_ids = match self
1045 .register_services(
1046 manifest,
1047 plugin_dir,
1048 &previously_owned_services,
1049 &mut rollback,
1050 )
1051 .await
1052 {
1053 Ok(ids) => ids,
1054 Err(error) => {
1055 self.abort_install(&rollback, &previous, &manifest.id, &installed_json_path)
1056 .await;
1057 return Err(error);
1058 }
1059 };
1060
1061 // Step 2: Prompts.
1062 let preset_ids = match self.register_prompts(manifest).await {
1063 Ok(ids) => {
1064 rollback.preset_ids_added = ids.clone();
1065 ids
1066 }
1067 Err(error) => {
1068 self.abort_install(&rollback, &previous, &manifest.id, &installed_json_path)
1069 .await;
1070 return Err(error);
1071 }
1072 };
1073
1074 // Step 3: Workflows. `register_workflows` records each copied file
1075 // into `rollback` itself as it goes (see its doc comment) — a
1076 // partial failure partway through a multi-file copy is still fully
1077 // rolled back.
1078 let workflow_filenames = match self
1079 .register_workflows(
1080 manifest,
1081 plugin_dir,
1082 &previously_owned_workflows,
1083 &mut rollback,
1084 )
1085 .await
1086 {
1087 Ok(files) => files,
1088 Err(error) => {
1089 self.abort_install(&rollback, &previous, &manifest.id, &installed_json_path)
1090 .await;
1091 return Err(error);
1092 }
1093 };
1094
1095 // Step 4: Skills — nothing to register, just record the
1096 // declared+validated dir names (preflight_install already confirmed
1097 // every declared dir exists and that no undeclared dir is present).
1098 let skill_dirs = manifest.provides.skills.clone();
1099
1100 // Step 5: commit provenance — flip the journal row to `Installed` with
1101 // the ACTUAL registered set (renamed preset ids, the to_register mcp/
1102 // workflow subsets). Only reached once 0-4 all succeeded.
1103 let registered = RegisteredCapabilities {
1104 mcp_server_ids,
1105 skill_dirs,
1106 preset_ids,
1107 workflow_filenames,
1108 service_ids,
1109 };
1110 let entry = InstalledPlugin {
1111 id: manifest.id.clone(),
1112 version: manifest.version.clone(),
1113 source,
1114 plugin_dir: plugin_dir.to_path_buf(),
1115 installed_at,
1116 status: PluginInstallStatus::Installed,
1117 registered,
1118 };
1119 self.upsert_provenance(entry.clone(), &installed_json_path)
1120 .await?;
1121
1122 Ok(entry)
1123 }
1124
1125 async fn uninstall(&self, id: &str) -> PluginResult<()> {
1126 // Serialize against every other plugin op (see module docs).
1127 let _op_guard = PLUGIN_OP_LOCK.lock().await;
1128
1129 let installed_json_path = self.installed_json_path();
1130 let mut store = InstalledPlugins::load(&installed_json_path).await?;
1131 // Works on an `Installing` (crash-leftover) row too, so a crashed
1132 // install is never un-uninstallable.
1133 let Some(entry) = store.get(id).cloned() else {
1134 return Err(PluginError::NotFound(id.to_string()));
1135 };
1136
1137 // De-register everything this plugin's `registered` set names — by
1138 // construction (see bamboo-plugin's ownership contract) this can
1139 // only ever be entries the plugin itself created. Idempotent: a
1140 // manually-removed entry is logged and skipped, never a hard error.
1141 self.deregister_capabilities(&entry.registered).await;
1142
1143 // Remove the plugin's own files BEFORE clearing provenance: if this
1144 // fails (e.g. a permission error), provenance is left intact so a
1145 // retry is safe (the de-registration above is idempotent, so
1146 // re-running it is a harmless no-op) rather than leaving an
1147 // unregistered-but-still-on-disk `skills/` dir that discovery would
1148 // keep picking up despite `uninstall` having "succeeded".
1149 match fs::remove_dir_all(&entry.plugin_dir).await {
1150 Ok(()) => {}
1151 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
1152 Err(error) => return Err(PluginError::Io(error)),
1153 }
1154
1155 store.remove(id);
1156 store.save(&installed_json_path).await?;
1157 Ok(())
1158 }
1159
1160 async fn list(&self) -> PluginResult<Vec<InstalledPlugin>> {
1161 let store = InstalledPlugins::load(&self.installed_json_path()).await?;
1162 Ok(store.plugins)
1163 }
1164}
1165
1166// ---------------------------------------------------------------------
1167// Free helpers shared between `ServerPluginInstaller` (instance methods
1168// above, which delegate here) and `boot_reconcile_services` (which has no
1169// `ServerPluginInstaller`/`AppState` handle to call instance methods on —
1170// see `app_state::builder`, which calls it before `AppState` finishes
1171// constructing).
1172// ---------------------------------------------------------------------
1173
1174/// See `ServerPluginInstaller::service_config_path`'s doc comment for the
1175/// full rationale (kept there since that's the reader's first encounter).
1176fn service_config_path_under(app_data_dir: &Path, plugin_id: &str) -> PathBuf {
1177 app_data_dir
1178 .join("plugin_service_config")
1179 .join(plugin_id)
1180 .join("config.json")
1181}
1182
1183fn resolve_service_config_under(
1184 app_data_dir: &Path,
1185 plugin_id: &str,
1186 entry: &ServiceManifestEntry,
1187 plugin_dir: &Path,
1188 platform: Platform,
1189) -> ServiceRuntimeConfig {
1190 let resolved = entry.resolve(plugin_dir, plugin_id, platform);
1191 ServiceRuntimeConfig {
1192 id: resolved.id,
1193 plugin_id: plugin_id.to_string(),
1194 name: resolved.name,
1195 command: resolved.command,
1196 args: resolved.args,
1197 cwd: resolved.cwd,
1198 env: resolved.env,
1199 health_check: resolved.health_check,
1200 restart_policy: resolved.restart_policy,
1201 graceful_shutdown: resolved.graceful_shutdown,
1202 user_config_path: service_config_path_under(app_data_dir, plugin_id),
1203 }
1204}
1205
1206/// Boot-time reconcile (issue #479): start every ENABLED, plugin-owned
1207/// service that `installed.json` says should be running but has no live
1208/// [`ServiceManager`] runtime — the previous `bamboo serve` process (if any)
1209/// died along with every service it supervised (child processes are spawned
1210/// `kill_on_drop`, and nothing about a running service persists
1211/// cross-process). Called from `app_state::builder` the same way
1212/// `app_state::init::init_mcp_manager` kicks off its background MCP
1213/// bootstrap — the caller is expected to `tokio::spawn` this, NOT await it
1214/// inline, so server startup is never blocked on plugin service spawns.
1215///
1216/// Deliberately reads `installed.json` + each plugin's on-disk
1217/// `plugin.json` directly rather than going through `ServerPluginInstaller`
1218/// (which needs a fully-built `web::Data<AppState>` this runs before).
1219pub async fn boot_reconcile_services(app_data_dir: &Path, service_manager: &ServiceManager) {
1220 let installed_json_path = app_data_dir.join("plugins").join("installed.json");
1221 let store = match InstalledPlugins::load(&installed_json_path).await {
1222 Ok(store) => store,
1223 Err(error) => {
1224 tracing::warn!(
1225 %error,
1226 "service boot-reconcile: failed to load installed.json; skipping"
1227 );
1228 return;
1229 }
1230 };
1231
1232 let platform = Platform::current().unwrap_or(Platform::Linux);
1233 for plugin in store.list() {
1234 if plugin.registered.service_ids.is_empty() {
1235 continue;
1236 }
1237 let manifest_path = plugin.plugin_dir.join("plugin.json");
1238 let manifest = match fs::read_to_string(&manifest_path)
1239 .await
1240 .ok()
1241 .and_then(|raw| PluginManifest::parse_str(&raw).ok())
1242 {
1243 Some(manifest) => manifest,
1244 None => {
1245 tracing::warn!(
1246 plugin_id = %plugin.id,
1247 path = %manifest_path.display(),
1248 "service boot-reconcile: failed to read/parse plugin.json; skipping this \
1249 plugin's services"
1250 );
1251 continue;
1252 }
1253 };
1254
1255 let owned: HashSet<&str> = plugin
1256 .registered
1257 .service_ids
1258 .iter()
1259 .map(String::as_str)
1260 .collect();
1261 for entry in &manifest.provides.services {
1262 if !entry.enabled || !owned.contains(entry.id.as_str()) {
1263 continue;
1264 }
1265 if service_manager.is_running(&entry.id) {
1266 continue;
1267 }
1268 let config = resolve_service_config_under(
1269 app_data_dir,
1270 &plugin.id,
1271 entry,
1272 &plugin.plugin_dir,
1273 platform,
1274 );
1275 if let Some(parent) = config.user_config_path.parent() {
1276 if let Err(error) = fs::create_dir_all(parent).await {
1277 tracing::warn!(
1278 service_id = %entry.id,
1279 plugin_id = %plugin.id,
1280 %error,
1281 "service boot-reconcile: failed to create service config parent dir"
1282 );
1283 }
1284 }
1285 match service_manager.start_service(config).await {
1286 Ok(()) => tracing::info!(
1287 service_id = %entry.id,
1288 plugin_id = %plugin.id,
1289 "service boot-reconcile: started"
1290 ),
1291 Err(error) => tracing::warn!(
1292 service_id = %entry.id,
1293 plugin_id = %plugin.id,
1294 %error,
1295 "service boot-reconcile: failed to start"
1296 ),
1297 }
1298 }
1299 }
1300}
1301
1302#[cfg(test)]
1303mod tests;