feat(backend): système de plugins — domaine, application, infrastructure (#43)
Lots B1-B4 : modèle de domaine des plugins (manifeste, capacités menus/layouts/MCP), port et registre applicatif, chargement/validation en infrastructure, exposition DTO et commandes Tauri. Tests cargo verts. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@ -30,6 +30,7 @@ pub mod model_server;
|
||||
pub mod orchestrator;
|
||||
pub mod pair_attempt_limiter;
|
||||
pub mod permission;
|
||||
pub mod plugin;
|
||||
pub mod process;
|
||||
pub mod pty;
|
||||
pub mod ratelimit;
|
||||
@ -80,6 +81,7 @@ pub use orchestrator::{
|
||||
};
|
||||
pub use pair_attempt_limiter::InMemoryPairAttemptLimiter;
|
||||
pub use permission::{ClaudePermissionProjector, CodexPermissionProjector};
|
||||
pub use plugin::{ExternalMcpPluginSupervisor, FsPluginPackageStore, FsPluginRegistryStore};
|
||||
pub use process::LocalProcessSpawner;
|
||||
pub use pty::PortablePtyAdapter;
|
||||
pub use ratelimit::RateLimitParser;
|
||||
|
||||
755
crates/infrastructure/src/plugin/mod.rs
Normal file
755
crates/infrastructure/src/plugin/mod.rs
Normal file
@ -0,0 +1,755 @@
|
||||
//! 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<PathBuf>) -> 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<PathBuf, PluginStoreError> {
|
||||
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<PathBuf, PluginStoreError> {
|
||||
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<Vec<PluginPackageRef>, 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<PluginManifestBytes, PluginStoreError> {
|
||||
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<StagedPluginPackage, PluginStoreError> {
|
||||
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<StagedPluginPackage, PluginStoreError> {
|
||||
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<PluginPackageRef, PluginStoreError> {
|
||||
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<RemovalOutcome, PluginStoreError> {
|
||||
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<PluginBundleUrl, PluginStoreError> {
|
||||
Ok(PluginBundleUrl::new(format!(
|
||||
"idea-plugin://{}/current/{}{}{}",
|
||||
plugin_id.as_str(),
|
||||
hash.as_str(),
|
||||
"/",
|
||||
entry.as_str()
|
||||
)))
|
||||
}
|
||||
|
||||
fn app_data_dir_label(&self) -> Option<String> {
|
||||
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<ContentHash, PluginStoreError> {
|
||||
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<PathBuf>) -> 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<PathBuf>) -> 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<PluginRegistry, PluginRegistryError> {
|
||||
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<Box<dyn ExternalMcpServerHandle>, 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<Box<dyn ExternalMcpServerHandle>, 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<dyn ExternalMcpServerBridge>,
|
||||
children: Mutex<HashMap<String, Box<dyn ExternalMcpServerHandle>>>,
|
||||
}
|
||||
|
||||
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<dyn ExternalMcpServerBridge>) -> 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<PluginMcpServerSpec>,
|
||||
) -> Result<PluginMcpStatusSet, PluginMcpError> {
|
||||
let desired: HashSet<String> = 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::<Vec<_>>()
|
||||
};
|
||||
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::<Vec<_>>()
|
||||
};
|
||||
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<Vec<PluginMcpServerSpec>>,
|
||||
stopped: Arc<Mutex<Vec<String>>>,
|
||||
}
|
||||
|
||||
struct RecordingHandle {
|
||||
identity: String,
|
||||
stopped: Arc<Mutex<Vec<String>>>,
|
||||
}
|
||||
|
||||
#[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<Box<dyn ExternalMcpServerHandle>, 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);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user