use std::collections::HashSet;
use std::path::{Path, PathBuf};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use tokio::fs;
use bamboo_domain::mcp_config::McpServerConfig;
use bamboo_plugin::installer::{load_previous_for_disposition, preflight_install};
use bamboo_plugin::manifest::{Platform, ServiceManifestEntry};
use bamboo_plugin::registry::{reconcile_exclusive, RegisteredCapabilities};
use bamboo_plugin::{
InstallDisposition, InstalledPlugin, InstalledPlugins, PluginError, PluginInstallStatus,
PluginInstaller, PluginManifest, PluginResult, PluginSource,
};
use crate::app_state::{AppState, ConfigUpdateEffects};
use crate::error::AppError;
use crate::handlers::agent::mcp::upsert_server_by_id;
use crate::handlers::agent::prompt_presets::{
ensure_unique_preset_id, load_store, save_store, store_file_path, StoredPromptPreset,
};
use crate::service_manager::{ServiceManager, ServiceRuntimeConfig};
static PLUGIN_OP_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
pub struct ServerPluginInstaller {
state: actix_web::web::Data<AppState>,
}
#[derive(Default)]
struct InstallRollback {
mcp_ids_added: Vec<String>,
mcp_ids_started: Vec<String>,
preset_ids_added: Vec<String>,
workflow_files_added: Vec<String>,
service_ids_added: Vec<String>,
service_ids_started: Vec<String>,
}
impl ServerPluginInstaller {
pub fn new(state: actix_web::web::Data<AppState>) -> Self {
Self { state }
}
fn plugins_dir(&self) -> PathBuf {
self.state.app_data_dir.join("plugins")
}
fn installed_json_path(&self) -> PathBuf {
self.plugins_dir().join("installed.json")
}
fn workflows_dir(&self) -> PathBuf {
self.state.app_data_dir.join("workflows")
}
fn prompt_presets_path(&self) -> PathBuf {
store_file_path(&self.state.app_data_dir)
}
fn service_config_path(&self, plugin_id: &str) -> PathBuf {
service_config_path_under(&self.state.app_data_dir, plugin_id)
}
async fn ensure_service_config_parent_dir(&self, plugin_id: &str) -> PluginResult<()> {
let path = self.service_config_path(plugin_id);
if let Some(parent) = path.parent() {
fs::create_dir_all(parent).await?;
}
Ok(())
}
async fn existing_service_ids(&self, exclude_plugin_id: &str) -> PluginResult<Vec<String>> {
let store = InstalledPlugins::load(&self.installed_json_path()).await?;
Ok(store
.list()
.iter()
.filter(|plugin| plugin.id != exclude_plugin_id)
.flat_map(|plugin| plugin.registered.service_ids.iter().cloned())
.collect())
}
fn resolve_service_config(
&self,
plugin_id: &str,
entry: &ServiceManifestEntry,
plugin_dir: &Path,
platform: Platform,
) -> ServiceRuntimeConfig {
resolve_service_config_under(
&self.state.app_data_dir,
plugin_id,
entry,
plugin_dir,
platform,
)
}
async fn existing_workflow_filenames(&self) -> PluginResult<Vec<String>> {
let dir = self.workflows_dir();
fs::create_dir_all(&dir).await?;
let mut entries = fs::read_dir(&dir).await?;
let mut names = Vec::new();
while let Some(entry) = entries.next_entry().await? {
let is_file = entry
.file_type()
.await
.map(|file_type| file_type.is_file())
.unwrap_or(false);
if !is_file {
continue;
}
if let Some(name) = entry.file_name().to_str() {
if name.ends_with(".md") {
names.push(name.to_string());
}
}
}
Ok(names)
}
async fn remove_mcp_server(&self, id: &str) {
let owned_id = id.to_string();
let result = self
.state
.update_config(
move |cfg| {
cfg.mcp.servers.retain(|server| server.id != owned_id);
Ok(())
},
ConfigUpdateEffects::default(),
)
.await;
if let Err(error) = result {
tracing::warn!(
mcp_server_id = %id,
%error,
"failed to remove plugin-owned mcp server from config.json; continuing"
);
}
if let Err(error) = self.state.mcp_manager.stop_server(id).await {
tracing::warn!(
mcp_server_id = %id,
%error,
"failed to stop plugin-owned mcp server; continuing"
);
}
}
async fn remove_prompt_preset(&self, preset_id: &str) {
let path = self.prompt_presets_path();
match load_store(&path).await {
Ok(mut store) => {
let before = store.prompts.len();
store.prompts.retain(|preset| preset.id != preset_id);
if store.prompts.len() != before {
if let Err(error) = save_store(&path, &store).await {
tracing::warn!(
%preset_id,
%error,
"failed to persist prompt-presets.json after removing plugin-owned preset; continuing"
);
}
}
}
Err(error) => {
tracing::warn!(
%preset_id,
%error,
"failed to load prompt-presets.json while removing plugin-owned preset; continuing"
);
}
}
}
async fn remove_service(&self, id: &str) {
if let Err(error) = self.state.service_manager.stop_service(id).await {
tracing::warn!(
service_id = %id,
%error,
"failed to stop plugin-owned service; continuing"
);
}
}
async fn remove_workflow_file(&self, filename: &str) {
let path = self.workflows_dir().join(filename);
match fs::remove_file(&path).await {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
tracing::warn!(
%filename,
%error,
"failed to remove plugin-owned workflow file; continuing"
);
}
}
}
async fn deregister_capabilities(&self, registered: &RegisteredCapabilities) {
for mcp_id in ®istered.mcp_server_ids {
self.remove_mcp_server(mcp_id).await;
}
for preset_id in ®istered.preset_ids {
self.remove_prompt_preset(preset_id).await;
}
for workflow_filename in ®istered.workflow_filenames {
self.remove_workflow_file(workflow_filename).await;
}
for service_id in ®istered.service_ids {
self.remove_service(service_id).await;
}
}
async fn rollback_partial_install(&self, rollback: &InstallRollback) {
for id in &rollback.mcp_ids_started {
let _ = self.state.mcp_manager.stop_server(id).await;
}
for id in &rollback.mcp_ids_added {
self.remove_mcp_server(id).await;
}
for id in &rollback.preset_ids_added {
self.remove_prompt_preset(id).await;
}
for file in &rollback.workflow_files_added {
self.remove_workflow_file(file).await;
}
for id in &rollback.service_ids_started {
let _ = self.state.service_manager.stop_service(id).await;
}
}
async fn upsert_provenance(&self, entry: InstalledPlugin, path: &Path) -> PluginResult<()> {
let mut store = InstalledPlugins::load(path).await?;
store.add(entry);
store.save(path).await?;
Ok(())
}
async fn abort_install(
&self,
rollback: &InstallRollback,
previous: &Option<InstalledPlugin>,
plugin_id: &str,
path: &Path,
) {
self.rollback_partial_install(rollback).await;
let restore = match previous {
Some(prev) => self.upsert_provenance(prev.clone(), path).await,
None => match InstalledPlugins::load(path).await {
Ok(mut store) => {
store.remove(plugin_id);
store.save(path).await
}
Err(error) => Err(error),
},
};
if let Err(error) = restore {
tracing::warn!(
%plugin_id,
%error,
"failed to restore provenance after aborting a failed install; a stale \
`installing` row may remain (recoverable by a retry)"
);
}
}
async fn register_mcp(
&self,
manifest: &PluginManifest,
resolved_mcp_servers: Vec<McpServerConfig>,
previously_owned: &[String],
rollback: &mut InstallRollback,
) -> PluginResult<Vec<String>> {
if resolved_mcp_servers.is_empty() {
return Ok(Vec::new());
}
let declared_ids: Vec<String> = manifest
.provides
.mcp_servers
.iter()
.map(|entry| entry.id.clone())
.collect();
let existing_ids: Vec<String> = {
let config = self.state.config.read().await;
config.mcp.servers.iter().map(|s| s.id.clone()).collect()
};
let reconciliation = reconcile_exclusive(&declared_ids, &existing_ids, previously_owned);
if !reconciliation.foreign_conflicts.is_empty() {
return Err(PluginError::Conflict {
kind: "mcp server",
name: reconciliation.foreign_conflicts.join(", "),
plugin_id: manifest.id.clone(),
});
}
let to_register: HashSet<&str> = reconciliation
.to_register
.iter()
.map(String::as_str)
.collect();
let configs_to_register: Vec<McpServerConfig> = resolved_mcp_servers
.into_iter()
.filter(|config| to_register.contains(config.id.as_str()))
.collect();
let owned_configs = configs_to_register.clone();
let declared_for_recheck = declared_ids.clone();
let owned_for_recheck: Vec<String> = previously_owned.to_vec();
let plugin_id_for_recheck = manifest.id.clone();
self.state
.update_config(
move |cfg| {
let live_existing: Vec<String> =
cfg.mcp.servers.iter().map(|s| s.id.clone()).collect();
let live = reconcile_exclusive(
&declared_for_recheck,
&live_existing,
&owned_for_recheck,
);
if !live.foreign_conflicts.is_empty() {
return Err(AppError::BadRequest(format!(
"mcp server(s) '{}' now conflict with a non-plugin entry (a concurrent \
change landed mid-install); refusing to overwrite for plugin '{}'",
live.foreign_conflicts.join(", "),
plugin_id_for_recheck
)));
}
for server in &owned_configs {
upsert_server_by_id(&mut cfg.mcp.servers, server.clone());
}
Ok(())
},
ConfigUpdateEffects::default(),
)
.await
.map_err(|error| {
PluginError::Registration(format!("failed to write mcp servers to config: {error}"))
})?;
rollback.mcp_ids_added = reconciliation.to_register.clone();
for server in &configs_to_register {
let _ = self.state.mcp_manager.stop_server(&server.id).await;
if server.enabled {
match self.state.mcp_manager.start_server(server.clone()).await {
Ok(()) => rollback.mcp_ids_started.push(server.id.clone()),
Err(error) => tracing::warn!(
mcp_server_id = %server.id,
%error,
"plugin-registered mcp server failed to start; config entry kept (best-effort)"
),
}
}
}
Ok(reconciliation.to_register)
}
async fn register_services(
&self,
manifest: &PluginManifest,
plugin_dir: &Path,
previously_owned: &[String],
rollback: &mut InstallRollback,
) -> PluginResult<Vec<String>> {
if manifest.provides.services.is_empty() {
return Ok(Vec::new());
}
let declared_ids: Vec<String> = manifest
.provides
.services
.iter()
.map(|entry| entry.id.clone())
.collect();
let existing_ids = self.existing_service_ids(&manifest.id).await?;
let reconciliation = reconcile_exclusive(&declared_ids, &existing_ids, previously_owned);
if !reconciliation.foreign_conflicts.is_empty() {
return Err(PluginError::Conflict {
kind: "service",
name: reconciliation.foreign_conflicts.join(", "),
plugin_id: manifest.id.clone(),
});
}
rollback.service_ids_added = reconciliation.to_register.clone();
self.ensure_service_config_parent_dir(&manifest.id).await?;
let platform = Platform::current().unwrap_or(Platform::Linux);
let to_register: HashSet<&str> = reconciliation
.to_register
.iter()
.map(String::as_str)
.collect();
for entry in &manifest.provides.services {
if !to_register.contains(entry.id.as_str()) {
continue;
}
let _ = self.state.service_manager.stop_service(&entry.id).await;
if !entry.enabled {
continue;
}
let config = self.resolve_service_config(&manifest.id, entry, plugin_dir, platform);
match self.state.service_manager.start_service(config).await {
Ok(()) => rollback.service_ids_started.push(entry.id.clone()),
Err(error) => tracing::warn!(
service_id = %entry.id,
%error,
"plugin-registered service failed to start; ownership kept (best-effort, matches mcp)"
),
}
}
Ok(reconciliation.to_register)
}
async fn register_prompts(&self, manifest: &PluginManifest) -> PluginResult<Vec<String>> {
if manifest.provides.prompts.is_empty() {
return Ok(Vec::new());
}
let path = self.prompt_presets_path();
let mut store = load_store(&path).await.map_err(|error| {
PluginError::Registration(format!("failed to load prompt-presets.json: {error}"))
})?;
let mut existing_ids: HashSet<String> = store
.prompts
.iter()
.map(|preset| preset.id.clone())
.collect();
let mut actual_ids = Vec::with_capacity(manifest.provides.prompts.len());
for preset in &manifest.provides.prompts {
let actual_id = ensure_unique_preset_id(&preset.id, &existing_ids);
store.prompts.push(StoredPromptPreset {
id: actual_id.clone(),
name: preset.name.clone(),
description: preset.description.clone(),
content: preset.content.clone(),
});
existing_ids.insert(actual_id.clone());
actual_ids.push(actual_id);
}
save_store(&path, &store).await.map_err(|error| {
PluginError::Registration(format!("failed to persist prompt-presets.json: {error}"))
})?;
Ok(actual_ids)
}
async fn register_workflows(
&self,
manifest: &PluginManifest,
plugin_dir: &Path,
previously_owned: &[String],
rollback: &mut InstallRollback,
) -> PluginResult<Vec<String>> {
if manifest.provides.workflows.is_empty() {
return Ok(Vec::new());
}
let declared: Vec<String> = manifest.provides.workflows.clone();
let existing = self.existing_workflow_filenames().await?;
let reconciliation = reconcile_exclusive(&declared, &existing, previously_owned);
if !reconciliation.foreign_conflicts.is_empty() {
return Err(PluginError::Conflict {
kind: "workflow",
name: reconciliation.foreign_conflicts.join(", "),
plugin_id: manifest.id.clone(),
});
}
let dest_dir = self.workflows_dir();
for filename in &reconciliation.to_register {
let stem = filename.strip_suffix(".md").unwrap_or(filename);
if !bamboo_config::paths::is_safe_workflow_name(stem) {
return Err(PluginError::InvalidManifest(format!(
"workflow filename '{filename}' is not a safe workflow name"
)));
}
let source_path = plugin_dir.join("workflows").join(filename);
let content = fs::read_to_string(&source_path).await?;
fs::write(dest_dir.join(filename), content).await?;
rollback.workflow_files_added.push(filename.clone());
}
Ok(reconciliation.to_register)
}
pub async fn stop_services_for_upgrade(&self, plugin_id: &str) -> Vec<String> {
let store = match InstalledPlugins::load(&self.installed_json_path()).await {
Ok(store) => store,
Err(error) => {
tracing::warn!(
%plugin_id,
%error,
"stop_services_for_upgrade: failed to load installed.json; skipping"
);
return Vec::new();
}
};
let Some(entry) = store.get(plugin_id) else {
return Vec::new();
};
let mut stopped = Vec::with_capacity(entry.registered.service_ids.len());
for service_id in &entry.registered.service_ids {
match self.state.service_manager.stop_service(service_id).await {
Ok(()) => stopped.push(service_id.clone()),
Err(error) => tracing::debug!(
service_id = %service_id,
%error,
"stop_services_for_upgrade: service was not running; nothing to stop"
),
}
}
stopped
}
pub async fn restart_services_after_failed_upgrade(&self, plugin_id: &str, stopped: &[String]) {
if stopped.is_empty() {
return;
}
let store = match InstalledPlugins::load(&self.installed_json_path()).await {
Ok(store) => store,
Err(error) => {
tracing::warn!(
%plugin_id,
%error,
"restart_services_after_failed_upgrade: failed to load installed.json"
);
return;
}
};
let Some(entry) = store.get(plugin_id) else {
return;
};
let manifest_path = entry.plugin_dir.join("plugin.json");
let manifest = match fs::read_to_string(&manifest_path)
.await
.ok()
.and_then(|raw| PluginManifest::parse_str(&raw).ok())
{
Some(manifest) => manifest,
None => {
tracing::warn!(
%plugin_id,
path = %manifest_path.display(),
"restart_services_after_failed_upgrade: failed to read/parse the \
rolled-back plugin.json; affected service(s) remain stopped"
);
return;
}
};
let platform = Platform::current().unwrap_or(Platform::Linux);
for svc in &manifest.provides.services {
if !stopped.contains(&svc.id) || !svc.enabled {
continue;
}
let config = self.resolve_service_config(plugin_id, svc, &entry.plugin_dir, platform);
if let Err(error) = self.state.service_manager.start_service(config).await {
tracing::warn!(
service_id = %svc.id,
%plugin_id,
%error,
"failed to restart service after a failed upgrade rolled back to the \
previous plugin bundle; service remains stopped"
);
}
}
}
}
#[async_trait]
impl PluginInstaller for ServerPluginInstaller {
async fn install(
&self,
manifest: &PluginManifest,
plugin_dir: &Path,
source: PluginSource,
disposition: InstallDisposition,
installed_at: DateTime<Utc>,
) -> PluginResult<InstalledPlugin> {
let _op_guard = PLUGIN_OP_LOCK.lock().await;
let installed_json_path = self.installed_json_path();
let previous =
load_previous_for_disposition(&installed_json_path, &manifest.id, disposition).await?;
let resolved_mcp_servers = preflight_install(manifest, plugin_dir).await?;
let intended = RegisteredCapabilities {
mcp_server_ids: manifest
.provides
.mcp_servers
.iter()
.map(|entry| entry.id.clone())
.collect(),
skill_dirs: manifest.provides.skills.clone(),
preset_ids: manifest
.provides
.prompts
.iter()
.map(|preset| preset.id.clone())
.collect(),
workflow_filenames: manifest.provides.workflows.clone(),
service_ids: manifest
.provides
.services
.iter()
.map(|entry| entry.id.clone())
.collect(),
};
if let Some(previous) = &previous {
let dropped = intended.removed_since(&previous.registered);
if !dropped.is_empty() {
tracing::info!(
plugin_id = %manifest.id,
recovering = previous.status == PluginInstallStatus::Installing,
dropped_mcp = ?dropped.mcp_server_ids,
dropped_presets = ?dropped.preset_ids,
dropped_workflows = ?dropped.workflow_filenames,
dropped_services = ?dropped.service_ids,
"install drop-diff: de-registering capabilities the new/completed version no longer declares"
);
self.deregister_capabilities(&dropped).await;
}
}
let previously_owned_mcp = previous
.as_ref()
.map(|p| p.registered.mcp_server_ids.clone())
.unwrap_or_default();
let previously_owned_workflows = previous
.as_ref()
.map(|p| p.registered.workflow_filenames.clone())
.unwrap_or_default();
let previously_owned_services = previous
.as_ref()
.map(|p| p.registered.service_ids.clone())
.unwrap_or_default();
self.upsert_provenance(
InstalledPlugin {
id: manifest.id.clone(),
version: manifest.version.clone(),
source: source.clone(),
plugin_dir: plugin_dir.to_path_buf(),
installed_at,
status: PluginInstallStatus::Installing,
registered: intended.clone(),
},
&installed_json_path,
)
.await?;
let mut rollback = InstallRollback::default();
let mcp_server_ids = match self
.register_mcp(
manifest,
resolved_mcp_servers,
&previously_owned_mcp,
&mut rollback,
)
.await
{
Ok(ids) => ids,
Err(error) => {
self.abort_install(&rollback, &previous, &manifest.id, &installed_json_path)
.await;
return Err(error);
}
};
let service_ids = match self
.register_services(
manifest,
plugin_dir,
&previously_owned_services,
&mut rollback,
)
.await
{
Ok(ids) => ids,
Err(error) => {
self.abort_install(&rollback, &previous, &manifest.id, &installed_json_path)
.await;
return Err(error);
}
};
let preset_ids = match self.register_prompts(manifest).await {
Ok(ids) => {
rollback.preset_ids_added = ids.clone();
ids
}
Err(error) => {
self.abort_install(&rollback, &previous, &manifest.id, &installed_json_path)
.await;
return Err(error);
}
};
let workflow_filenames = match self
.register_workflows(
manifest,
plugin_dir,
&previously_owned_workflows,
&mut rollback,
)
.await
{
Ok(files) => files,
Err(error) => {
self.abort_install(&rollback, &previous, &manifest.id, &installed_json_path)
.await;
return Err(error);
}
};
let skill_dirs = manifest.provides.skills.clone();
let registered = RegisteredCapabilities {
mcp_server_ids,
skill_dirs,
preset_ids,
workflow_filenames,
service_ids,
};
let entry = InstalledPlugin {
id: manifest.id.clone(),
version: manifest.version.clone(),
source,
plugin_dir: plugin_dir.to_path_buf(),
installed_at,
status: PluginInstallStatus::Installed,
registered,
};
self.upsert_provenance(entry.clone(), &installed_json_path)
.await?;
Ok(entry)
}
async fn uninstall(&self, id: &str) -> PluginResult<()> {
let _op_guard = PLUGIN_OP_LOCK.lock().await;
let installed_json_path = self.installed_json_path();
let mut store = InstalledPlugins::load(&installed_json_path).await?;
let Some(entry) = store.get(id).cloned() else {
return Err(PluginError::NotFound(id.to_string()));
};
self.deregister_capabilities(&entry.registered).await;
match fs::remove_dir_all(&entry.plugin_dir).await {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(PluginError::Io(error)),
}
store.remove(id);
store.save(&installed_json_path).await?;
Ok(())
}
async fn list(&self) -> PluginResult<Vec<InstalledPlugin>> {
let store = InstalledPlugins::load(&self.installed_json_path()).await?;
Ok(store.plugins)
}
}
fn service_config_path_under(app_data_dir: &Path, plugin_id: &str) -> PathBuf {
app_data_dir
.join("plugin_service_config")
.join(plugin_id)
.join("config.json")
}
fn resolve_service_config_under(
app_data_dir: &Path,
plugin_id: &str,
entry: &ServiceManifestEntry,
plugin_dir: &Path,
platform: Platform,
) -> ServiceRuntimeConfig {
let resolved = entry.resolve(plugin_dir, plugin_id, platform);
ServiceRuntimeConfig {
id: resolved.id,
plugin_id: plugin_id.to_string(),
name: resolved.name,
command: resolved.command,
args: resolved.args,
cwd: resolved.cwd,
env: resolved.env,
health_check: resolved.health_check,
restart_policy: resolved.restart_policy,
graceful_shutdown: resolved.graceful_shutdown,
user_config_path: service_config_path_under(app_data_dir, plugin_id),
}
}
pub async fn boot_reconcile_services(app_data_dir: &Path, service_manager: &ServiceManager) {
let installed_json_path = app_data_dir.join("plugins").join("installed.json");
let store = match InstalledPlugins::load(&installed_json_path).await {
Ok(store) => store,
Err(error) => {
tracing::warn!(
%error,
"service boot-reconcile: failed to load installed.json; skipping"
);
return;
}
};
let platform = Platform::current().unwrap_or(Platform::Linux);
for plugin in store.list() {
if plugin.registered.service_ids.is_empty() {
continue;
}
let manifest_path = plugin.plugin_dir.join("plugin.json");
let manifest = match fs::read_to_string(&manifest_path)
.await
.ok()
.and_then(|raw| PluginManifest::parse_str(&raw).ok())
{
Some(manifest) => manifest,
None => {
tracing::warn!(
plugin_id = %plugin.id,
path = %manifest_path.display(),
"service boot-reconcile: failed to read/parse plugin.json; skipping this \
plugin's services"
);
continue;
}
};
let owned: HashSet<&str> = plugin
.registered
.service_ids
.iter()
.map(String::as_str)
.collect();
for entry in &manifest.provides.services {
if !entry.enabled || !owned.contains(entry.id.as_str()) {
continue;
}
if service_manager.is_running(&entry.id) {
continue;
}
let config = resolve_service_config_under(
app_data_dir,
&plugin.id,
entry,
&plugin.plugin_dir,
platform,
);
if let Some(parent) = config.user_config_path.parent() {
if let Err(error) = fs::create_dir_all(parent).await {
tracing::warn!(
service_id = %entry.id,
plugin_id = %plugin.id,
%error,
"service boot-reconcile: failed to create service config parent dir"
);
}
}
match service_manager.start_service(config).await {
Ok(()) => tracing::info!(
service_id = %entry.id,
plugin_id = %plugin.id,
"service boot-reconcile: started"
),
Err(error) => tracing::warn!(
service_id = %entry.id,
plugin_id = %plugin.id,
%error,
"service boot-reconcile: failed to start"
),
}
}
}
}
#[cfg(test)]
mod tests;