//! Filesystem plugin stores and external MCP supervisor. use std::collections::{BTreeMap, HashMap, HashSet}; use std::fs; use std::io::Read; use std::path::{Path, PathBuf}; use std::process::Stdio; use std::sync::{Arc, Mutex}; use std::time::{SystemTime, UNIX_EPOCH}; use async_trait::async_trait; use domain::ports::{ LocalPath, PluginManifestBytes, PluginMcpError, PluginMcpSupervisor, PluginPackageStore, PluginRegistryError, PluginRegistryStore, PluginStoreError, }; use domain::{ ContentHash, PluginBundleUrl, PluginId, PluginInstallSource, PluginMcpServerSpec, PluginMcpStatus, PluginMcpStatusSet, PluginPackageRef, PluginRegistry, RelativePath, RemovalOutcome, StagedPluginPackage, }; use sha2::{Digest, Sha256}; use tokio::process::Child; const REGISTRY_FILE: &str = "registry.json"; const MANIFEST_FILE: &str = "idea-plugin.json"; /// Filesystem package store under app-data `plugins/`. #[derive(Debug, Clone)] pub struct FsPluginPackageStore { root: PathBuf, } impl FsPluginPackageStore { /// Builds the store. #[must_use] pub fn new(app_data_dir: impl Into) -> Self { Self { root: app_data_dir.into().join("plugins"), } } fn installed_dir(&self) -> PathBuf { self.root.join("installed") } /// Returns the global plugin store root. #[must_use] pub fn plugins_root(&self) -> PathBuf { self.root.clone() } /// Returns the managed directory for one installed plugin id. #[must_use] pub fn installed_plugin_dir(&self, plugin_id: &PluginId) -> PathBuf { self.installed_dir().join(plugin_id.as_str()) } fn staging_dir(&self) -> PathBuf { self.root.join("_staging") } fn trash_dir(&self) -> PathBuf { self.root.join("_trash") } fn package_root(&self, package: &PluginPackageRef) -> PathBuf { match &package.plugin_id { Some(id) => self.installed_dir().join(id.as_str()), None => PathBuf::from(&package.root), } } /// Resolves an asset path if the request is root-confined. pub fn resolve_asset_path( &self, plugin_id: &PluginId, path: &RelativePath, ) -> Result { let root = self.installed_dir().join(plugin_id.as_str()); let candidate = root.join(path.as_str()); let canonical_root = root .canonicalize() .map_err(|e| PluginStoreError::Io(e.to_string()))?; let canonical = candidate .canonicalize() .map_err(|e| PluginStoreError::Io(e.to_string()))?; if canonical.starts_with(canonical_root) { Ok(canonical) } else { Err(PluginStoreError::Invalid( "plugin asset escapes package root".to_owned(), )) } } fn stage_root(&self) -> Result { let ts = SystemTime::now() .duration_since(UNIX_EPOCH) .map_err(|e| PluginStoreError::Io(e.to_string()))? .as_nanos(); let path = self.staging_dir().join(format!("stage-{ts}")); fs::create_dir_all(&path).map_err(|e| PluginStoreError::Io(e.to_string()))?; Ok(path) } } #[async_trait] impl PluginPackageStore for FsPluginPackageStore { async fn list_installed(&self) -> Result, PluginStoreError> { let dir = self.installed_dir(); if !dir.exists() { return Ok(Vec::new()); } let mut out = Vec::new(); for entry in fs::read_dir(dir).map_err(|e| PluginStoreError::Io(e.to_string()))? { let entry = entry.map_err(|e| PluginStoreError::Io(e.to_string()))?; if !entry .file_type() .map_err(|e| PluginStoreError::Io(e.to_string()))? .is_dir() { continue; } let name = entry.file_name().to_string_lossy().into_owned(); if let Ok(id) = PluginId::new(name) { out.push(PluginPackageRef { plugin_id: Some(id), root: entry.path().to_string_lossy().into_owned(), }); } } Ok(out) } async fn read_manifest( &self, package: &PluginPackageRef, ) -> Result { let bytes = fs::read(self.package_root(package).join(MANIFEST_FILE)).map_err(|e| { if e.kind() == std::io::ErrorKind::NotFound { PluginStoreError::NotFound } else { PluginStoreError::Io(e.to_string()) } })?; Ok(PluginManifestBytes { bytes }) } async fn install_from_archive( &self, archive: &LocalPath, ) -> Result { let stage = self.stage_root()?; let status = std::process::Command::new("unzip") .arg("-q") .arg(archive.as_str()) .arg("-d") .arg(&stage) .status() .map_err(|e| PluginStoreError::Io(format!("failed to run unzip: {e}")))?; if !status.success() { return Err(PluginStoreError::Format(format!( "unzip exited with status {status}" ))); } ensure_manifest(&stage)?; let content_hash = hash_dir(&stage)?; Ok(StagedPluginPackage { root: stage.to_string_lossy().into_owned(), source: PluginInstallSource::Archive { path_label: archive.as_str().to_owned(), }, content_hash, }) } async fn install_from_directory( &self, dir: &LocalPath, ) -> Result { let source = PathBuf::from(dir.as_str()); if !source.is_dir() { return Err(PluginStoreError::NotFound); } let stage = self.stage_root()?; copy_dir_all(&source, &stage)?; ensure_manifest(&stage)?; let content_hash = hash_dir(&stage)?; Ok(StagedPluginPackage { root: stage.to_string_lossy().into_owned(), source: PluginInstallSource::Directory { path_label: dir.as_str().to_owned(), }, content_hash, }) } async fn commit_install( &self, staged: StagedPluginPackage, plugin_id: &PluginId, ) -> Result { fs::create_dir_all(self.installed_dir()) .map_err(|e| PluginStoreError::Io(e.to_string()))?; let target = self.installed_dir().join(plugin_id.as_str()); let tmp = self .installed_dir() .join(format!(".{}-new", plugin_id.as_str())); if tmp.exists() { fs::remove_dir_all(&tmp).map_err(|e| PluginStoreError::Io(e.to_string()))?; } fs::rename(&staged.root, &tmp).or_else(|_| { copy_dir_all(Path::new(&staged.root), &tmp)?; fs::remove_dir_all(&staged.root).map_err(|e| PluginStoreError::Io(e.to_string())) })?; if target.exists() { fs::remove_dir_all(&target).map_err(|e| PluginStoreError::Io(e.to_string()))?; } fs::rename(&tmp, &target).map_err(|e| PluginStoreError::Io(e.to_string()))?; Ok(PluginPackageRef { plugin_id: Some(plugin_id.clone()), root: target.to_string_lossy().into_owned(), }) } async fn remove_package( &self, plugin_id: &PluginId, ) -> Result { let target = self.installed_dir().join(plugin_id.as_str()); if !target.exists() { return Ok(RemovalOutcome::NotFound); } match fs::remove_dir_all(&target) { Ok(()) => Ok(RemovalOutcome::Removed), Err(_) => { fs::create_dir_all(self.trash_dir()) .map_err(|e| PluginStoreError::Io(e.to_string()))?; let ts = SystemTime::now() .duration_since(UNIX_EPOCH) .map_err(|e| PluginStoreError::Io(e.to_string()))? .as_secs(); let tombstone = self .trash_dir() .join(format!("{}-{ts}", plugin_id.as_str())); fs::rename(&target, tombstone).map_err(|e| PluginStoreError::Io(e.to_string()))?; Ok(RemovalOutcome::Tombstoned) } } } fn bundle_url( &self, plugin_id: &PluginId, entry: &RelativePath, hash: &ContentHash, ) -> Result { Ok(PluginBundleUrl::new(format!( "idea-plugin://{}/current/{}{}{}", plugin_id.as_str(), hash.as_str(), "/", entry.as_str() ))) } fn app_data_dir_label(&self) -> Option { self.root .parent() .map(|p| p.to_string_lossy().into_owned()) } } fn ensure_manifest(root: &Path) -> Result<(), PluginStoreError> { if root.join(MANIFEST_FILE).is_file() { Ok(()) } else { Err(PluginStoreError::NotFound) } } fn copy_dir_all(source: &Path, target: &Path) -> Result<(), PluginStoreError> { fs::create_dir_all(target).map_err(|e| PluginStoreError::Io(e.to_string()))?; for entry in fs::read_dir(source).map_err(|e| PluginStoreError::Io(e.to_string()))? { let entry = entry.map_err(|e| PluginStoreError::Io(e.to_string()))?; let ty = entry .file_type() .map_err(|e| PluginStoreError::Io(e.to_string()))?; let dest = target.join(entry.file_name()); if ty.is_dir() { copy_dir_all(&entry.path(), &dest)?; } else if ty.is_file() { fs::copy(entry.path(), dest).map_err(|e| PluginStoreError::Io(e.to_string()))?; } } Ok(()) } fn hash_dir(root: &Path) -> Result { let mut files = Vec::new(); collect_files(root, &mut files)?; files.sort(); let mut hasher = Sha256::new(); for path in files { let rel = path .strip_prefix(root) .map_err(|e| PluginStoreError::Io(e.to_string()))? .to_string_lossy() .replace('\\', "/"); hasher.update(rel.as_bytes()); hasher.update([0]); let mut file = fs::File::open(&path).map_err(|e| PluginStoreError::Io(e.to_string()))?; let mut buf = Vec::new(); file.read_to_end(&mut buf) .map_err(|e| PluginStoreError::Io(e.to_string()))?; hasher.update(buf); hasher.update([0]); } ContentHash::new(hex::encode(hasher.finalize())) .map_err(|e| PluginStoreError::Invalid(e.to_string())) } fn collect_files(root: &Path, files: &mut Vec) -> Result<(), PluginStoreError> { for entry in fs::read_dir(root).map_err(|e| PluginStoreError::Io(e.to_string()))? { let entry = entry.map_err(|e| PluginStoreError::Io(e.to_string()))?; let ty = entry .file_type() .map_err(|e| PluginStoreError::Io(e.to_string()))?; if ty.is_dir() { collect_files(&entry.path(), files)?; } else if ty.is_file() { files.push(entry.path()); } } Ok(()) } /// Filesystem registry store. #[derive(Debug, Clone)] pub struct FsPluginRegistryStore { root: PathBuf, } impl FsPluginRegistryStore { /// Builds the store. #[must_use] pub fn new(app_data_dir: impl Into) -> Self { Self { root: app_data_dir.into().join("plugins"), } } fn path(&self) -> PathBuf { self.root.join(REGISTRY_FILE) } } #[async_trait] impl PluginRegistryStore for FsPluginRegistryStore { async fn load_registry(&self) -> Result { match fs::read(self.path()) { Ok(bytes) => serde_json::from_slice(&bytes) .map_err(|e| PluginRegistryError::Serialization(e.to_string())), Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(PluginRegistry::default()), Err(e) => Err(PluginRegistryError::Io(e.to_string())), } } async fn save_registry(&self, registry: &PluginRegistry) -> Result<(), PluginRegistryError> { fs::create_dir_all(&self.root).map_err(|e| PluginRegistryError::Io(e.to_string()))?; let bytes = serde_json::to_vec_pretty(registry) .map_err(|e| PluginRegistryError::Serialization(e.to_string()))?; fs::write(self.path(), bytes).map_err(|e| PluginRegistryError::Io(e.to_string())) } } #[async_trait] trait ExternalMcpServerHandle: Send { async fn stop(&mut self) -> Result<(), PluginMcpError>; } #[async_trait] trait ExternalMcpServerBridge: Send + Sync { async fn start( &self, spec: &PluginMcpServerSpec, ) -> Result, PluginMcpError>; } struct ProcessMcpServerHandle { child: Child, } #[async_trait] impl ExternalMcpServerHandle for ProcessMcpServerHandle { async fn stop(&mut self) -> Result<(), PluginMcpError> { self.child .kill() .await .map_err(|e| PluginMcpError::Process(e.to_string())) } } #[derive(Debug, Default)] struct StdioExternalMcpServerBridge; #[async_trait] impl ExternalMcpServerBridge for StdioExternalMcpServerBridge { async fn start( &self, spec: &PluginMcpServerSpec, ) -> Result, PluginMcpError> { if spec.transport != "stdio" { return Err(PluginMcpError::Process(format!( "unsupported plugin MCP transport: {}", spec.transport ))); } let mut cmd = tokio::process::Command::new(&spec.command); cmd.args(&spec.args) .current_dir(&spec.cwd) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()); let env: BTreeMap<_, _> = spec.env.iter().cloned().collect(); cmd.envs(env); let child = cmd .spawn() .map_err(|e| PluginMcpError::Process(e.to_string()))?; Ok(Box::new(ProcessMcpServerHandle { child })) } } /// External process supervisor for plugin MCP servers. pub struct ExternalMcpPluginSupervisor { bridge: Arc, children: Mutex>>, } impl ExternalMcpPluginSupervisor { /// Builds the supervisor. #[must_use] pub fn new() -> Self { Self { bridge: Arc::new(StdioExternalMcpServerBridge), children: Mutex::new(HashMap::new()), } } #[cfg(test)] fn with_bridge(bridge: Arc) -> Self { Self { bridge, children: Mutex::new(HashMap::new()), } } } impl Default for ExternalMcpPluginSupervisor { fn default() -> Self { Self::new() } } #[async_trait] impl PluginMcpSupervisor for ExternalMcpPluginSupervisor { async fn reconcile( &self, active_servers: Vec, ) -> Result { let desired: HashSet = active_servers.iter().map(|s| s.identity.clone()).collect(); let to_stop = { let children = self .children .lock() .expect("plugin mcp supervisor poisoned"); children .keys() .filter(|id| !desired.contains(*id)) .cloned() .collect::>() }; for id in to_stop { let child = self .children .lock() .expect("plugin mcp supervisor poisoned") .remove(&id); if let Some(mut child) = child { let _ = child.stop().await; } } let mut statuses = Vec::new(); for spec in active_servers { let already = self .children .lock() .expect("plugin mcp supervisor poisoned") .contains_key(&spec.identity); if already { statuses.push(PluginMcpStatus { identity: spec.identity, running: true, error: None, }); continue; } match self.bridge.start(&spec).await { Ok(handle) => { self.children .lock() .expect("plugin mcp supervisor poisoned") .insert(spec.identity.clone(), handle); statuses.push(PluginMcpStatus { identity: spec.identity, running: true, error: None, }); } Err(e) => statuses.push(PluginMcpStatus { identity: spec.identity, running: false, error: Some(e.to_string()), }), } } Ok(PluginMcpStatusSet { servers: statuses }) } async fn stop_plugin(&self, plugin_id: &PluginId) -> Result<(), PluginMcpError> { let prefix = format!("plugin:{}:", plugin_id.as_str()); let ids = { let children = self .children .lock() .expect("plugin mcp supervisor poisoned"); children .keys() .filter(|id| id.starts_with(&prefix)) .cloned() .collect::>() }; for id in ids { let child = self .children .lock() .expect("plugin mcp supervisor poisoned") .remove(&id); if let Some(mut child) = child { child.stop().await?; } } Ok(()) } } #[cfg(test)] mod tests { use super::*; use domain::ports::PluginPackageStore; use std::sync::atomic::{AtomicU64, Ordering}; static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); fn temp_dir(label: &str) -> PathBuf { let n = TEMP_COUNTER.fetch_add(1, Ordering::SeqCst); let path = std::env::temp_dir().join(format!( "idea-plugin-test-{label}-{}-{n}", std::process::id() )); let _ = fs::remove_dir_all(&path); fs::create_dir_all(&path).unwrap(); path } fn write_plugin(root: &Path, main_body: &str) { fs::create_dir_all(root.join("dist")).unwrap(); fs::write(root.join("dist/index.js"), main_body).unwrap(); fs::write( root.join(MANIFEST_FILE), r#"{"ideaPluginManifestVersion":1,"id":"dev.acme.test","displayName":"Test","version":"1.0.0","main":"dist/index.js","trustLevel":"full","contributes":{}}"#, ) .unwrap(); } #[derive(Default)] struct RecordingBridge { started: Mutex>, stopped: Arc>>, } struct RecordingHandle { identity: String, stopped: Arc>>, } #[async_trait] impl ExternalMcpServerHandle for RecordingHandle { async fn stop(&mut self) -> Result<(), PluginMcpError> { self.stopped.lock().unwrap().push(self.identity.clone()); Ok(()) } } #[async_trait] impl ExternalMcpServerBridge for RecordingBridge { async fn start( &self, spec: &PluginMcpServerSpec, ) -> Result, PluginMcpError> { assert_eq!(spec.transport, "stdio"); self.started.lock().unwrap().push(spec.clone()); Ok(Box::new(RecordingHandle { identity: spec.identity.clone(), stopped: Arc::clone(&self.stopped), })) } } fn mcp_spec(plugin_id: &str, server_id: &str) -> PluginMcpServerSpec { let plugin_id = PluginId::new(plugin_id).unwrap(); let server_id = domain::PluginMcpServerId::new(server_id).unwrap(); PluginMcpServerSpec { identity: format!("plugin:{}:{}", plugin_id.as_str(), server_id.as_str()), plugin_id, server_id, display_name: "Plugin Tools".to_owned(), command: "/plugin/servers/tool".to_owned(), args: vec!["--stdio".to_owned()], env: vec![("PLUGIN_ROOT".to_owned(), "/plugin".to_owned())], cwd: "/plugin".to_owned(), transport: "stdio".to_owned(), } } #[tokio::test] async fn installs_directory_snapshot_and_uninstalls_cleanly() { let app = temp_dir("app"); let source = temp_dir("source"); write_plugin(&source, "one"); let store = FsPluginPackageStore::new(&app); let staged = store .install_from_directory(&LocalPath::new(source.to_string_lossy())) .await .unwrap(); assert_ne!(staged.root, source.to_string_lossy()); let id = PluginId::new("dev.acme.test").unwrap(); let package = store.commit_install(staged, &id).await.unwrap(); assert!(PathBuf::from(package.root).join(MANIFEST_FILE).exists()); assert_eq!( store.remove_package(&id).await.unwrap(), RemovalOutcome::Removed ); assert!(!app.join("plugins/installed/dev.acme.test").exists()); let _ = fs::remove_dir_all(app); let _ = fs::remove_dir_all(source); } #[tokio::test] async fn content_hash_changes_when_bundle_changes() { let app = temp_dir("hash-app"); let source = temp_dir("hash-source"); write_plugin(&source, "one"); let store = FsPluginPackageStore::new(&app); let first = store .install_from_directory(&LocalPath::new(source.to_string_lossy())) .await .unwrap() .content_hash; fs::write(source.join("dist/index.js"), "two").unwrap(); let second = store .install_from_directory(&LocalPath::new(source.to_string_lossy())) .await .unwrap() .content_hash; assert_ne!(first, second); let _ = fs::remove_dir_all(app); let _ = fs::remove_dir_all(source); } #[tokio::test] async fn extracts_archive_without_path_escape() { if std::process::Command::new("zip") .arg("-h") .stdout(Stdio::null()) .stderr(Stdio::null()) .status() .is_err() || std::process::Command::new("unzip") .arg("-h") .stdout(Stdio::null()) .stderr(Stdio::null()) .status() .is_err() { return; } let app = temp_dir("archive-app"); let source = temp_dir("archive-source"); write_plugin(&source, "bundle"); let archive_path = app.join("plugin.ideaplug"); { let status = std::process::Command::new("zip") .arg("-qr") .arg(&archive_path) .arg(".") .current_dir(&source) .status() .unwrap(); assert!(status.success()); } let store = FsPluginPackageStore::new(app.join("data")); let staged = store .install_from_archive(&LocalPath::new(archive_path.to_string_lossy())) .await .unwrap(); assert!(PathBuf::from(staged.root).join(MANIFEST_FILE).exists()); let _ = fs::remove_dir_all(app); let _ = fs::remove_dir_all(source); } #[tokio::test] async fn supervisor_delegates_stdio_servers_to_external_mcp_bridge() { let bridge = Arc::new(RecordingBridge::default()); let supervisor = ExternalMcpPluginSupervisor::with_bridge(bridge.clone()); let spec = mcp_spec("dev.acme.gitgraph", "dev.acme.gitgraph.mcp"); let statuses = supervisor.reconcile(vec![spec.clone()]).await.unwrap(); assert_eq!(statuses.servers.len(), 1); assert_eq!(statuses.servers[0].identity, spec.identity); assert!(statuses.servers[0].running); assert_eq!(&*bridge.started.lock().unwrap(), &[spec]); } #[tokio::test] async fn supervisor_stop_plugin_stops_only_matching_plugin_servers() { let bridge = Arc::new(RecordingBridge::default()); let supervisor = ExternalMcpPluginSupervisor::with_bridge(bridge.clone()); let target = mcp_spec("dev.acme.gitgraph", "dev.acme.gitgraph.mcp"); let other = mcp_spec("dev.other.tools", "dev.other.tools.mcp"); supervisor .reconcile(vec![target.clone(), other.clone()]) .await .unwrap(); supervisor.stop_plugin(&target.plugin_id).await.unwrap(); assert_eq!(&*bridge.stopped.lock().unwrap(), &[target.identity.clone()]); let statuses = supervisor.reconcile(vec![other.clone()]).await.unwrap(); assert_eq!(statuses.servers.len(), 1); assert_eq!(statuses.servers[0].identity, other.identity); assert_eq!(bridge.started.lock().unwrap().len(), 2); } }