Files
IdeA/crates/application/src/plugin/mod.rs
Blomios e2da2d911e feat(plugins): load activation scope from plugin manifest
Plugins can now declare activationScope ("app" | "project") in their
manifest; loader/runtime honor it to defer activation of project-scoped
plugins until a project is focused instead of activating everything at
app bootstrap. Bumps sdk/IdeaSDK to the commit that adds the field.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-03 14:44:08 +02:00

5035 lines
167 KiB
Rust

//! Plugin application use cases.
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::{Arc, Mutex};
use domain::ports::EnvironmentReader;
use domain::ports::{
BackgroundTaskStore, Clock, DirEntry, EventBus, FileMetadata, FileSystem, IdGenerator,
LocalPath, Output, PluginManifestBytes, PluginManifestError, PluginManifestValidator,
PluginMcpError, PluginMcpSupervisor, PluginPackageStore, PluginRegistryError,
PluginRegistryStore, PluginStorageError, PluginStorageStore, PluginStoreError, ProcessError,
ProcessSpawner, ProjectStore, RemotePath, SpawnSpec,
};
use domain::{
AgentId, BackgroundTask, BackgroundTaskState, BackgroundTaskWakePolicy, ContentHash,
DomainEvent, PluginContributionSet, PluginDescriptor, PluginId, PluginInstallSource,
PluginLifecycleState, PluginManifest, PluginMcpServerSpec, PluginRegistryEntry,
PluginTrustLevel, Project, ProjectId, ProjectPath, RemovalOutcome, StagedPluginPackage, TaskId,
};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::{AppError, SpawnBackgroundCommand, SpawnBackgroundCommandInput};
/// Contribution counts for admin display.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginContributionSummary {
/// Top-level menu count.
pub top_level_menus: usize,
/// Menu item count.
pub menu_items: usize,
/// Layout count.
pub layouts: usize,
/// MCP server count.
pub mcp_servers: usize,
}
impl From<&PluginContributionSet> for PluginContributionSummary {
fn from(c: &PluginContributionSet) -> Self {
Self {
top_level_menus: c.menus.len(),
menu_items: c.menu_items.len(),
layouts: c.layouts.len(),
mcp_servers: c.mcp_servers.len(),
}
}
}
/// Admin plugin view.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginAdmin {
/// Plugin id.
pub id: String,
/// Display name.
pub display_name: String,
/// Publisher.
pub publisher: Option<String>,
/// Version.
pub version: String,
/// Description.
pub description: Option<String>,
/// Icon URL.
pub icon_url: Option<String>,
/// Source kind.
pub source_kind: String,
/// Source label.
pub source_label: Option<String>,
/// Lifecycle state.
pub lifecycle_state: PluginLifecycleState,
/// Enabled projection.
pub enabled: bool,
/// Pending enable state.
pub pending_enable_state: Option<bool>,
/// Pending uninstall flag.
pub pending_uninstall: bool,
/// Restart required flag.
pub restart_required: bool,
/// Trust level.
pub trust_level: PluginTrustLevel,
/// Summary.
pub contribution_summary: PluginContributionSummary,
/// Error.
pub error: Option<String>,
}
/// Pre-install package review.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginReview {
/// Manifest.
pub manifest: PluginManifest,
/// Source.
pub source: PluginInstallSource,
/// Content hash.
pub content_hash: String,
/// Summary.
pub contribution_summary: PluginContributionSummary,
/// Full-trust marker.
pub trust_level: PluginTrustLevel,
}
/// Install result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginInstallResult {
/// Installed plugin.
pub plugin: PluginAdmin,
/// Review used for installation.
pub review: PluginReview,
/// Restart required flag.
pub restart_required: bool,
}
/// Uninstall result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UninstallPluginResult {
/// Plugin id.
pub plugin_id: String,
/// Removal outcome.
pub removal_outcome: RemovalOutcome,
/// Restart required flag.
pub restart_required: bool,
}
/// Runtime catalog.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginRuntimeCatalog {
/// Runtime plugins.
pub plugins: Vec<PluginRuntimePlugin>,
}
/// Runtime plugin entry.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginRuntimePlugin {
/// Plugin id.
pub id: String,
/// Display name.
pub display_name: String,
/// Publisher.
pub publisher: Option<String>,
/// Version.
pub version: String,
/// Bundle URL.
pub bundle_url: String,
/// Icon URL.
pub icon_url: Option<String>,
/// Content hash.
pub content_hash: String,
/// Public manifest capabilities.
pub capabilities: Vec<domain::PluginCapability>,
/// Manifest-declared activation scope.
pub activation_scope: domain::PluginActivationScope,
/// Contributions.
pub contributes: PluginContributionSet,
}
/// Input for plugin-owned storage reads/deletes.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginStorageGetInput {
/// Plugin id owning the value.
pub plugin_id: String,
/// Plugin-owned key.
pub key: String,
}
/// Input for plugin-owned storage writes.
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginStorageSetInput {
/// Plugin id owning the value.
pub plugin_id: String,
/// Plugin-owned key.
pub key: String,
/// JSON value to persist.
pub value: serde_json::Value,
}
/// Input for reviewing a package.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReviewPluginPackageInput {
/// Review local archive.
Archive {
/// Path.
path: String,
},
/// Review local directory.
Directory {
/// Path.
path: String,
},
}
/// Input used by plugin workspace file commands.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginWorkspacePathInput {
/// Project id owning the workspace root.
pub project_id: String,
/// Relative path inside the project root. Empty or `.` targets the root.
pub path: String,
}
/// Text write input for plugin workspace files.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginWorkspaceWriteTextInput {
/// Project id owning the workspace root.
pub project_id: String,
/// Relative path inside the project root.
pub path: String,
/// UTF-8 content to write.
pub content: String,
}
/// Binary write input for plugin workspace files.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginWorkspaceWriteBinaryInput {
/// Project id owning the workspace root.
pub project_id: String,
/// Relative path inside the project root.
pub path: String,
/// Raw bytes to write.
pub bytes: Vec<u8>,
}
/// Text file read result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginWorkspaceTextFile {
/// Normalized relative path.
pub path: String,
/// UTF-8 content.
pub content: String,
}
/// Binary file read result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginWorkspaceBinaryFile {
/// Normalized relative path.
pub path: String,
/// Raw bytes.
pub bytes: Vec<u8>,
}
/// Directory entry visible to plugins.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginWorkspaceDirEntry {
/// Entry basename.
pub name: String,
/// Normalized relative path from project root.
pub path: String,
/// Whether the entry is a directory.
pub is_dir: bool,
}
/// Directory listing result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginWorkspaceDirectoryListing {
/// Normalized relative path listed.
pub path: String,
/// Entries sorted by name for stable plugin behavior.
pub entries: Vec<PluginWorkspaceDirEntry>,
}
/// Basic stat result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginWorkspaceStat {
/// Normalized relative path.
pub path: String,
/// Whether the path exists.
pub exists: bool,
/// Whether the path is a file, when known.
pub is_file: bool,
/// Whether the path is a directory, when known.
pub is_dir: bool,
/// File length in bytes when known.
pub len: Option<u64>,
}
/// Safe resolved workspace path.
#[derive(Debug, Clone, PartialEq, Eq)]
struct ResolvedWorkspacePath {
relative: String,
absolute: RemotePath,
}
/// Generic workspace/file API exposed to plugins through public use cases.
pub struct PluginWorkspaceAccess {
projects: Arc<dyn ProjectStore>,
fs: Arc<dyn FileSystem>,
events: Option<Arc<dyn EventBus>>,
}
impl PluginWorkspaceAccess {
/// Builds the workspace access facade.
#[must_use]
pub fn new(projects: Arc<dyn ProjectStore>, fs: Arc<dyn FileSystem>) -> Self {
Self {
projects,
fs,
events: None,
}
}
/// Attaches a public workspace-change event publisher.
#[must_use]
pub fn with_events(mut self, events: Arc<dyn EventBus>) -> Self {
self.events = Some(events);
self
}
/// Reads a UTF-8 file under the project root.
///
/// # Errors
/// [`AppError`] for unknown projects, invalid paths, invalid UTF-8 or I/O failures.
pub async fn read_text(
&self,
input: PluginWorkspacePathInput,
) -> Result<PluginWorkspaceTextFile, AppError> {
let (_project, path) = self.resolve_input(&input).await?;
let bytes = self.fs.read(&path.absolute).await?;
let content = String::from_utf8(bytes)
.map_err(|_| AppError::Invalid(format!("file is not valid UTF-8: {}", input.path)))?;
Ok(PluginWorkspaceTextFile {
path: path.relative,
content,
})
}
/// Reads raw bytes under the project root.
///
/// # Errors
/// [`AppError`] for unknown projects, invalid paths or I/O failures.
pub async fn read_binary(
&self,
input: PluginWorkspacePathInput,
) -> Result<PluginWorkspaceBinaryFile, AppError> {
let (_project, path) = self.resolve_input(&input).await?;
let bytes = self.fs.read(&path.absolute).await?;
Ok(PluginWorkspaceBinaryFile {
path: path.relative,
bytes,
})
}
/// Writes UTF-8 content under the project root.
///
/// # Errors
/// [`AppError`] for unknown projects, invalid paths or I/O failures.
pub async fn write_text(&self, input: PluginWorkspaceWriteTextInput) -> Result<(), AppError> {
let path_input = PluginWorkspacePathInput {
project_id: input.project_id,
path: input.path,
};
let (project, path) = self.resolve_input(&path_input).await?;
self.fs
.write(&path.absolute, input.content.as_bytes())
.await?;
self.publish_workspace_file_changed(project.id, &path.relative);
Ok(())
}
/// Writes raw bytes under the project root.
///
/// # Errors
/// [`AppError`] for unknown projects, invalid paths or I/O failures.
pub async fn write_binary(
&self,
input: PluginWorkspaceWriteBinaryInput,
) -> Result<(), AppError> {
let path_input = PluginWorkspacePathInput {
project_id: input.project_id,
path: input.path,
};
let (project, path) = self.resolve_input(&path_input).await?;
self.fs.write(&path.absolute, &input.bytes).await?;
self.publish_workspace_file_changed(project.id, &path.relative);
Ok(())
}
/// Lists one directory under the project root.
///
/// # Errors
/// [`AppError`] for unknown projects, invalid paths or I/O failures.
pub async fn list_dir(
&self,
input: PluginWorkspacePathInput,
) -> Result<PluginWorkspaceDirectoryListing, AppError> {
let (_project, path) = self.resolve_input(&input).await?;
let mut entries = self.fs.list(&path.absolute).await?;
entries.sort_by(|a, b| a.name.cmp(&b.name));
Ok(PluginWorkspaceDirectoryListing {
path: path.relative.clone(),
entries: entries
.into_iter()
.map(|entry| dir_entry_to_workspace_entry(&path.relative, entry))
.collect(),
})
}
/// Stats one path under the project root.
///
/// # Errors
/// [`AppError`] for unknown projects, invalid paths or permission/I/O failures.
pub async fn stat(
&self,
input: PluginWorkspacePathInput,
) -> Result<PluginWorkspaceStat, AppError> {
let (_project, path) = self.resolve_input(&input).await?;
match self.fs.metadata(&path.absolute).await {
Ok(metadata) => Ok(stat_from_metadata(path.relative, true, metadata)),
Err(domain::ports::FsError::NotFound(_)) => Ok(PluginWorkspaceStat {
path: path.relative,
exists: false,
is_file: false,
is_dir: false,
len: None,
}),
Err(err) => Err(AppError::from(err)),
}
}
async fn resolve_input(
&self,
input: &PluginWorkspacePathInput,
) -> Result<(Project, ResolvedWorkspacePath), AppError> {
let project_id = parse_project_id(&input.project_id)?;
let project = self.projects.load_project(project_id).await?;
let path = resolve_workspace_path(&project, &input.path)?;
Ok((project, path))
}
fn publish_workspace_file_changed(&self, project_id: ProjectId, path: &str) {
if let Some(events) = &self.events {
events.publish(DomainEvent::PluginWorkspaceFileChanged {
project_id,
path: path.to_owned(),
operation: "changed".to_owned(),
});
}
}
}
fn parse_project_id(raw: &str) -> Result<ProjectId, AppError> {
Uuid::parse_str(raw)
.map(ProjectId::from_uuid)
.map_err(|_| AppError::Invalid(format!("invalid project id: {raw}")))
}
fn dir_entry_to_workspace_entry(parent: &str, entry: DirEntry) -> PluginWorkspaceDirEntry {
let path = if parent.is_empty() {
entry.name.clone()
} else {
format!("{parent}/{}", entry.name)
};
PluginWorkspaceDirEntry {
name: entry.name,
path,
is_dir: entry.is_dir,
}
}
fn stat_from_metadata(path: String, exists: bool, metadata: FileMetadata) -> PluginWorkspaceStat {
PluginWorkspaceStat {
path,
exists,
is_file: metadata.is_file,
is_dir: metadata.is_dir,
len: metadata.len,
}
}
fn resolve_workspace_path(
project: &Project,
raw_relative: &str,
) -> Result<ResolvedWorkspacePath, AppError> {
let relative = normalize_workspace_relative_path(raw_relative)?;
let root = project.root.as_str();
let absolute = if relative.is_empty() {
root.to_owned()
} else {
let separator = if root.contains('\\') && !root.contains('/') {
"\\"
} else {
"/"
};
format!(
"{}{}{}",
root.trim_end_matches(['/', '\\']),
separator,
relative.replace('/', separator)
)
};
Ok(ResolvedWorkspacePath {
relative,
absolute: RemotePath::new(absolute),
})
}
fn normalize_workspace_relative_path(raw: &str) -> Result<String, AppError> {
let raw = raw.trim();
if raw.is_empty() || raw == "." {
return Ok(String::new());
}
if raw.contains('\0') || raw.starts_with('/') || raw.starts_with('\\') || raw.contains(':') {
return Err(AppError::Invalid(format!(
"workspace path must be relative to the project root: {raw}"
)));
}
let normalized = raw.replace('\\', "/");
let mut parts = Vec::new();
for part in normalized.split('/') {
if part.is_empty() || part == "." || part == ".." {
return Err(AppError::Invalid(format!(
"workspace path must not contain empty, '.', or '..' segments: {raw}"
)));
}
parts.push(part);
}
Ok(parts.join("/"))
}
/// Input for reading one structured configuration document.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginConfigDocumentReadInput {
/// Project id owning the workspace root.
pub project_id: String,
/// Relative document path under the project root.
pub path: String,
/// Optional explicit format. When omitted, the format is inferred from the extension.
#[serde(default)]
pub format: Option<String>,
}
/// Input for updating one structured configuration document.
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginConfigDocumentUpdateInput {
/// Project id owning the workspace root.
pub project_id: String,
/// Relative document path under the project root.
pub path: String,
/// Optional explicit format. When omitted, the format is inferred from the extension.
#[serde(default)]
pub format: Option<String>,
/// Update mode: `mergePatch` (default) or `replace`.
#[serde(default)]
pub mode: Option<String>,
/// JSON value used as replacement or merge patch.
pub value: serde_json::Value,
}
/// Structured configuration document visible to plugins.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginConfigDocument {
/// Project id.
pub project_id: String,
/// Normalized relative document path.
pub path: String,
/// Document format.
pub format: String,
/// Parsed document value.
pub value: serde_json::Value,
}
/// Structured configuration document write result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginConfigDocumentWriteResult {
/// Project id.
pub project_id: String,
/// Normalized relative document path.
pub path: String,
/// Document format.
pub format: String,
/// Applied update mode.
pub mode: String,
/// Number of bytes written.
pub bytes_written: usize,
}
/// Public plugin facade for structured configuration documents.
pub struct PluginConfigDocuments {
projects: Arc<dyn ProjectStore>,
fs: Arc<dyn FileSystem>,
events: Option<Arc<dyn EventBus>>,
}
impl PluginConfigDocuments {
/// Builds the facade.
#[must_use]
pub fn new(projects: Arc<dyn ProjectStore>, fs: Arc<dyn FileSystem>) -> Self {
Self {
projects,
fs,
events: None,
}
}
/// Attaches a public workspace-change event publisher.
#[must_use]
pub fn with_events(mut self, events: Arc<dyn EventBus>) -> Self {
self.events = Some(events);
self
}
/// Reads one structured configuration document.
///
/// # Errors
/// [`AppError`] for unknown projects, invalid paths, unsupported formats, invalid UTF-8,
/// invalid document syntax, or I/O failures.
pub async fn read(
&self,
input: PluginConfigDocumentReadInput,
) -> Result<PluginConfigDocument, AppError> {
let (project, path) = self
.resolve_document(&input.project_id, &input.path)
.await?;
let format = resolve_config_format(input.format.as_deref(), &path.relative)?;
let bytes = self.fs.read(&path.absolute).await?;
let text = String::from_utf8(bytes).map_err(|_| {
AppError::Invalid(format!("document is not valid UTF-8: {}", input.path))
})?;
let value = parse_config_document(&format, &text)?;
Ok(PluginConfigDocument {
project_id: project.id.to_string(),
path: path.relative,
format,
value,
})
}
/// Updates one structured configuration document.
///
/// # Errors
/// [`AppError`] for unknown projects, invalid paths, unsupported formats, invalid UTF-8,
/// invalid document syntax, invalid update mode, or I/O failures.
pub async fn update(
&self,
input: PluginConfigDocumentUpdateInput,
) -> Result<PluginConfigDocumentWriteResult, AppError> {
let (project, path) = self
.resolve_document(&input.project_id, &input.path)
.await?;
let format = resolve_config_format(input.format.as_deref(), &path.relative)?;
let mode = normalize_config_update_mode(input.mode.as_deref())?;
let next = match mode.as_str() {
"replace" => input.value,
"mergePatch" => {
let bytes = self.fs.read(&path.absolute).await?;
let text = String::from_utf8(bytes).map_err(|_| {
AppError::Invalid(format!("document is not valid UTF-8: {}", input.path))
})?;
let mut current = parse_config_document(&format, &text)?;
apply_json_merge_patch(&mut current, input.value);
current
}
_ => unreachable!("mode is normalized"),
};
let text = serialize_config_document(&format, &next)?;
self.fs.write(&path.absolute, text.as_bytes()).await?;
self.publish_workspace_file_changed(project.id, &path.relative);
Ok(PluginConfigDocumentWriteResult {
project_id: project.id.to_string(),
path: path.relative,
format,
mode,
bytes_written: text.len(),
})
}
async fn resolve_document(
&self,
project_id: &str,
raw_path: &str,
) -> Result<(Project, ResolvedWorkspacePath), AppError> {
let project_id = parse_project_id(project_id)?;
let project = self.projects.load_project(project_id).await?;
let path = resolve_workspace_path(&project, raw_path)?;
Ok((project, path))
}
fn publish_workspace_file_changed(&self, project_id: ProjectId, path: &str) {
if let Some(events) = &self.events {
events.publish(DomainEvent::PluginWorkspaceFileChanged {
project_id,
path: path.to_owned(),
operation: "changed".to_owned(),
});
}
}
}
fn resolve_config_format(raw: Option<&str>, path: &str) -> Result<String, AppError> {
let format = match raw.map(str::trim).filter(|value| !value.is_empty()) {
Some(value) => value.to_ascii_lowercase(),
None if path.ends_with(".json") => "json".to_owned(),
None => {
return Err(AppError::Invalid(format!(
"could not infer structured config format for: {path}"
)))
}
};
if format == "json" {
Ok(format)
} else {
Err(AppError::Invalid(format!(
"unsupported structured config format: {format}; supported formats: json"
)))
}
}
fn normalize_config_update_mode(raw: Option<&str>) -> Result<String, AppError> {
match raw.map(str::trim).filter(|value| !value.is_empty()) {
None => Ok("mergePatch".to_owned()),
Some("mergePatch") | Some("replace") => Ok(raw.unwrap().trim().to_owned()),
Some(other) => Err(AppError::Invalid(format!(
"unsupported structured config update mode: {other}; supported modes: mergePatch, replace"
))),
}
}
fn parse_config_document(format: &str, text: &str) -> Result<serde_json::Value, AppError> {
match format {
"json" => serde_json::from_str(text)
.map_err(|err| AppError::Invalid(format!("invalid json document: {err}"))),
_ => Err(AppError::Invalid(format!(
"unsupported structured config format: {format}; supported formats: json"
))),
}
}
fn serialize_config_document(format: &str, value: &serde_json::Value) -> Result<String, AppError> {
match format {
"json" => {
let mut text = serde_json::to_string_pretty(value)
.map_err(|err| AppError::Invalid(format!("invalid json value: {err}")))?;
text.push('\n');
Ok(text)
}
_ => Err(AppError::Invalid(format!(
"unsupported structured config format: {format}; supported formats: json"
))),
}
}
fn apply_json_merge_patch(target: &mut serde_json::Value, patch: serde_json::Value) {
match patch {
serde_json::Value::Object(patch) => {
if !target.is_object() {
*target = serde_json::Value::Object(serde_json::Map::new());
}
let target = target.as_object_mut().expect("target object was just set");
for (key, value) in patch {
if value.is_null() {
target.remove(&key);
} else {
apply_json_merge_patch(
target.entry(key).or_insert(serde_json::Value::Null),
value,
);
}
}
}
value => *target = value,
}
}
/// Input for querying a bounded, generic project structure.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct QueryProjectStructureInput {
/// Project id owning the workspace root.
pub project_id: String,
/// Optional relative root to inspect.
#[serde(default)]
pub path: Option<String>,
/// Maximum directory depth to traverse. Defaults to 3 and is capped at 8.
#[serde(default)]
pub max_depth: Option<u8>,
/// Maximum number of entries to return. Defaults to 500 and is capped at 5000.
#[serde(default)]
pub max_entries: Option<usize>,
}
/// Bounded project structure query result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProjectStructureQuery {
/// Project id.
pub project_id: String,
/// Query root, relative to the project root.
pub root_path: String,
/// Returned entries.
pub entries: Vec<ProjectStructureEntry>,
/// Detected generic conventions.
pub conventions: Vec<ProjectConvention>,
/// Logical modules inferred from generic marker files.
pub modules: Vec<ProjectModule>,
/// Whether traversal stopped because `maxEntries` was reached.
pub truncated: bool,
}
/// One project structure entry.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProjectStructureEntry {
/// Relative path.
pub path: String,
/// Basename.
pub name: String,
/// Entry kind: `file` or `directory`.
pub kind: String,
}
/// Generic convention detected from marker files.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProjectConvention {
/// Stable convention id.
pub id: String,
/// Marker path that triggered the convention.
pub marker_path: String,
}
/// Logical project module inferred from marker files.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ProjectModule {
/// Relative module root path.
pub path: String,
/// Marker path that made this directory a module.
pub marker_path: String,
/// Convention id associated with the marker.
pub convention_id: String,
}
/// Generic project structure query use case for plugins.
pub struct QueryProjectStructure {
projects: Arc<dyn ProjectStore>,
fs: Arc<dyn FileSystem>,
}
impl QueryProjectStructure {
/// Builds the use case.
#[must_use]
pub fn new(projects: Arc<dyn ProjectStore>, fs: Arc<dyn FileSystem>) -> Self {
Self { projects, fs }
}
/// Executes a bounded structure query.
///
/// # Errors
/// [`AppError`] for unknown projects, invalid paths or I/O failures.
pub async fn execute(
&self,
input: QueryProjectStructureInput,
) -> Result<ProjectStructureQuery, AppError> {
let project_id = parse_project_id(&input.project_id)?;
let project = self.projects.load_project(project_id).await?;
let root_path = input.path.unwrap_or_default();
let root = resolve_workspace_path(&project, &root_path)?;
let max_depth = input.max_depth.unwrap_or(3).min(8);
let max_entries = input.max_entries.unwrap_or(500).min(5000);
let mut builder = StructureBuilder {
fs: self.fs.as_ref(),
entries: Vec::new(),
conventions: Vec::new(),
modules: Vec::new(),
truncated: false,
max_depth,
max_entries,
};
builder.visit_dir(&project, &root.relative, 0).await?;
Ok(ProjectStructureQuery {
project_id: input.project_id,
root_path: root.relative,
entries: builder.entries,
conventions: builder.conventions,
modules: builder.modules,
truncated: builder.truncated,
})
}
}
struct StructureBuilder<'a> {
fs: &'a dyn FileSystem,
entries: Vec<ProjectStructureEntry>,
conventions: Vec<ProjectConvention>,
modules: Vec<ProjectModule>,
truncated: bool,
max_depth: u8,
max_entries: usize,
}
impl StructureBuilder<'_> {
async fn visit_dir(
&mut self,
project: &Project,
relative: &str,
depth: u8,
) -> Result<(), AppError> {
if self.truncated || depth > self.max_depth {
return Ok(());
}
let resolved = resolve_workspace_path(project, relative)?;
let mut children = self.fs.list(&resolved.absolute).await?;
children.sort_by(|a, b| a.name.cmp(&b.name));
detect_module_markers(
relative,
&children,
&mut self.conventions,
&mut self.modules,
);
for child in children {
if self.entries.len() >= self.max_entries {
self.truncated = true;
return Ok(());
}
let child_path = if relative.is_empty() {
child.name.clone()
} else {
format!("{relative}/{}", child.name)
};
let kind = if child.is_dir { "directory" } else { "file" }.to_owned();
self.entries.push(ProjectStructureEntry {
path: child_path.clone(),
name: child.name,
kind,
});
if child.is_dir && depth < self.max_depth && should_descend(&child_path) {
Box::pin(self.visit_dir(project, &child_path, depth + 1)).await?;
}
}
Ok(())
}
}
fn should_descend(path: &str) -> bool {
let name = path.rsplit('/').next().unwrap_or(path);
!matches!(
name,
".git" | ".idea" | ".ideai" | "node_modules" | "target" | "dist" | "build"
)
}
fn detect_module_markers(
dir: &str,
children: &[DirEntry],
conventions: &mut Vec<ProjectConvention>,
modules: &mut Vec<ProjectModule>,
) {
for child in children.iter().filter(|entry| !entry.is_dir) {
if let Some(convention_id) = convention_for_marker(&child.name) {
let marker_path = if dir.is_empty() {
child.name.clone()
} else {
format!("{dir}/{}", child.name)
};
conventions.push(ProjectConvention {
id: convention_id.to_owned(),
marker_path: marker_path.clone(),
});
modules.push(ProjectModule {
path: dir.to_owned(),
marker_path,
convention_id: convention_id.to_owned(),
});
}
}
}
fn convention_for_marker(name: &str) -> Option<&'static str> {
match name {
"Cargo.toml" => Some("rust-cargo"),
"package.json" => Some("node-package"),
"pyproject.toml" | "setup.py" => Some("python-project"),
"go.mod" => Some("go-module"),
"pom.xml" => Some("maven-project"),
"Makefile" | "makefile" => Some("makefile"),
".git" => Some("git-repository"),
_ => None,
}
}
/// Input for launching a command-backed task from the public plugin API.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginRunCommandInput {
/// Project id owning the command workspace.
pub project_id: String,
/// Agent id that owns Work-panel correlation and optional wake delivery.
pub owner_agent_id: String,
/// Human-facing task label.
pub label: String,
/// Executable to run.
pub command: String,
/// Arguments passed without shell parsing.
#[serde(default)]
pub args: Vec<String>,
/// Optional relative working directory under the project root. Defaults to root.
#[serde(default)]
pub cwd: Option<String>,
/// Extra environment variables.
#[serde(default)]
pub env: Vec<(String, String)>,
/// When true, completion is only recorded; otherwise the owner is woken.
#[serde(default)]
pub record_only: bool,
/// Optional absolute deadline, epoch milliseconds.
#[serde(default)]
pub deadline_ms: Option<u64>,
}
/// Input for reading one plugin-launched task status.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginTaskStatusInput {
/// Task id to read.
pub task_id: String,
}
/// Toolchain diagnostic request for the public plugin API.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginToolchainDiagnosticInput {
/// Project id owning the workspace root.
pub project_id: String,
/// Relative working directory under the project root. Defaults to root.
#[serde(default)]
pub cwd: Option<String>,
/// Executable probes to run.
#[serde(default)]
pub tools: Vec<PluginToolRequirement>,
/// Environment variable prerequisites to read and validate.
#[serde(default)]
pub env: Vec<PluginEnvRequirement>,
/// Workspace file prerequisites to validate.
#[serde(default)]
pub files: Vec<PluginFileRequirement>,
}
/// Declarative executable probe requested by a plugin.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginToolRequirement {
/// Stable requirement id chosen by the plugin.
pub id: String,
/// Executable name or absolute path.
pub executable: String,
/// Arguments used to read a version or diagnostic. Defaults to `--version`.
#[serde(default)]
pub version_args: Vec<String>,
/// Whether this probe must pass for the whole diagnostic to be ok.
#[serde(default)]
pub required: bool,
/// Extra environment variables for this probe.
#[serde(default)]
pub env: Vec<(String, String)>,
}
/// Declarative environment variable prerequisite.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginEnvRequirement {
/// Environment variable name.
pub name: String,
/// Whether the variable must be present and match.
#[serde(default)]
pub required: bool,
/// Optional exact value requirement.
#[serde(default)]
pub equals: Option<String>,
}
/// Declarative workspace file prerequisite.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginFileRequirement {
/// Relative workspace path.
pub path: String,
/// Whether the path must exist and match `kind`.
#[serde(default)]
pub required: bool,
/// Optional kind: `file`, `directory`, or `any`.
#[serde(default)]
pub kind: Option<String>,
}
/// Toolchain diagnostic result for the public plugin API.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginToolchainDiagnostic {
/// Project id inspected.
pub project_id: String,
/// Working directory used for executable probes.
pub cwd: String,
/// Whether every required prerequisite passed.
pub ok: bool,
/// Executable probe results.
pub tools: Vec<PluginToolDiagnostic>,
/// Environment prerequisite results.
pub env: Vec<PluginEnvDiagnostic>,
/// File prerequisite results.
pub files: Vec<PluginFileDiagnostic>,
/// Human-readable diagnostics.
pub messages: Vec<PluginDiagnosticMessage>,
}
/// One executable probe result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginToolDiagnostic {
/// Requirement id.
pub id: String,
/// Executable name or path.
pub executable: String,
/// Whether the tool could be started.
pub present: bool,
/// Whether the probe satisfied this requirement.
pub ok: bool,
/// Probe status: `ok`, `failed`, or `missing`.
pub status: String,
/// Whether this requirement was required.
pub required: bool,
/// Process exit code, if the process started.
pub exit_code: Option<i32>,
/// First non-empty stdout/stderr line observed.
pub version: Option<String>,
/// Bounded stdout diagnostic.
pub stdout: Option<String>,
/// Bounded stderr diagnostic.
pub stderr: Option<String>,
/// Error text when the process could not start or run.
pub error: Option<String>,
}
/// One environment prerequisite result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginEnvDiagnostic {
/// Environment variable name.
pub name: String,
/// Whether the variable was present.
pub present: bool,
/// Whether the variable satisfied this requirement.
pub ok: bool,
/// Whether this requirement was required.
pub required: bool,
/// Observed value, if present.
pub value: Option<String>,
/// Environment status: `ok`, `missing`, or `mismatch`.
pub status: String,
}
/// One workspace file prerequisite result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginFileDiagnostic {
/// Normalized relative path.
pub path: String,
/// Whether the path exists.
pub exists: bool,
/// Whether the file prerequisite was satisfied.
pub ok: bool,
/// Whether this requirement was required.
pub required: bool,
/// Observed kind: `file`, `directory`, `other`, or `missing`.
pub kind: String,
/// Requested kind, if any.
pub expected_kind: Option<String>,
/// File length in bytes when known.
pub len: Option<u64>,
}
/// One human-readable diagnostic message.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginDiagnosticMessage {
/// Severity: `info`, `warning`, or `error`.
pub level: String,
/// Message text.
pub message: String,
}
/// Public plugin facade for generic external toolchain diagnostics.
pub struct PluginToolchainDiagnostics {
projects: Arc<dyn ProjectStore>,
fs: Arc<dyn FileSystem>,
processes: Arc<dyn ProcessSpawner>,
env: Arc<dyn EnvironmentReader>,
}
impl PluginToolchainDiagnostics {
/// Builds the facade.
#[must_use]
pub fn new(
projects: Arc<dyn ProjectStore>,
fs: Arc<dyn FileSystem>,
processes: Arc<dyn ProcessSpawner>,
env: Arc<dyn EnvironmentReader>,
) -> Self {
Self {
projects,
fs,
processes,
env,
}
}
/// Runs a generic diagnostic over executable, environment, and file requirements.
///
/// # Errors
/// [`AppError`] for malformed ids, unsafe workspace paths, unknown projects, or I/O failures.
pub async fn diagnose(
&self,
input: PluginToolchainDiagnosticInput,
) -> Result<PluginToolchainDiagnostic, AppError> {
let project_id = parse_project_id(&input.project_id)?;
let project = self.projects.load_project(project_id).await?;
let cwd_relative = input.cwd.as_deref().unwrap_or(".");
let cwd = resolve_workspace_path(&project, cwd_relative)?;
let cwd_path = ProjectPath::new(cwd.absolute.as_str().to_owned())
.map_err(|err| AppError::Invalid(err.to_string()))?;
let mut messages = Vec::new();
let mut ok = true;
let mut tools = Vec::new();
for requirement in input.tools {
let diagnostic = self
.diagnose_tool(requirement, cwd_path.clone(), &mut messages)
.await?;
if diagnostic.required && !diagnostic.ok {
ok = false;
}
tools.push(diagnostic);
}
let mut env = Vec::new();
for requirement in input.env {
let diagnostic = diagnose_env(self.env.as_ref(), requirement, &mut messages)?;
if diagnostic.required && !diagnostic.ok {
ok = false;
}
env.push(diagnostic);
}
let mut files = Vec::new();
for requirement in input.files {
let diagnostic = self
.diagnose_file(&project, requirement, &mut messages)
.await?;
if diagnostic.required && !diagnostic.ok {
ok = false;
}
files.push(diagnostic);
}
Ok(PluginToolchainDiagnostic {
project_id: input.project_id,
cwd: cwd.relative,
ok,
tools,
env,
files,
messages,
})
}
async fn diagnose_tool(
&self,
requirement: PluginToolRequirement,
cwd: ProjectPath,
messages: &mut Vec<PluginDiagnosticMessage>,
) -> Result<PluginToolDiagnostic, AppError> {
let id = trimmed_non_empty("tool id", &requirement.id)?;
let executable = trimmed_non_empty("tool executable", &requirement.executable)?;
let args = if requirement.version_args.is_empty() {
vec!["--version".to_owned()]
} else {
requirement.version_args
};
let output = self
.processes
.run(SpawnSpec {
command: executable.clone(),
args,
cwd,
env: requirement.env,
context_plan: None,
sandbox: None,
})
.await;
Ok(match output {
Ok(output) => diagnostic_from_output(id, executable, requirement.required, output),
Err(err) => {
let message = match &err {
ProcessError::Spawn(message) | ProcessError::Io(message) => message.clone(),
};
messages.push(PluginDiagnosticMessage {
level: if requirement.required {
"error".to_owned()
} else {
"warning".to_owned()
},
message: format!("{id}: {message}"),
});
PluginToolDiagnostic {
id,
executable,
present: false,
ok: false,
status: "missing".to_owned(),
required: requirement.required,
exit_code: None,
version: None,
stdout: None,
stderr: None,
error: Some(message),
}
}
})
}
async fn diagnose_file(
&self,
project: &Project,
requirement: PluginFileRequirement,
messages: &mut Vec<PluginDiagnosticMessage>,
) -> Result<PluginFileDiagnostic, AppError> {
let resolved = resolve_workspace_path(project, &requirement.path)?;
let expected_kind = normalize_expected_kind(requirement.kind)?;
match self.fs.metadata(&resolved.absolute).await {
Ok(metadata) => {
let kind = metadata_kind(&metadata);
let kind_ok = expected_kind
.as_deref()
.map_or(true, |expected| expected == "any" || expected == kind);
Ok(PluginFileDiagnostic {
path: resolved.relative,
exists: true,
ok: kind_ok,
required: requirement.required,
kind: kind.to_owned(),
expected_kind,
len: metadata.len,
})
}
Err(domain::ports::FsError::NotFound(_)) => {
if requirement.required {
messages.push(PluginDiagnosticMessage {
level: "error".to_owned(),
message: format!("missing required path: {}", resolved.relative),
});
}
Ok(PluginFileDiagnostic {
path: resolved.relative,
exists: false,
ok: false,
required: requirement.required,
kind: "missing".to_owned(),
expected_kind,
len: None,
})
}
Err(err) => Err(AppError::from(err)),
}
}
}
fn diagnostic_from_output(
id: String,
executable: String,
required: bool,
output: Output,
) -> PluginToolDiagnostic {
let exit_code = output.status.code;
let stdout = bounded_utf8(output.stdout);
let stderr = bounded_utf8(output.stderr);
let version =
first_non_empty_line(stdout.as_deref()).or_else(|| first_non_empty_line(stderr.as_deref()));
let ok = exit_code == Some(0);
PluginToolDiagnostic {
id,
executable,
present: true,
ok,
status: if ok { "ok" } else { "failed" }.to_owned(),
required,
exit_code,
version,
stdout,
stderr,
error: None,
}
}
fn diagnose_env(
reader: &dyn EnvironmentReader,
requirement: PluginEnvRequirement,
messages: &mut Vec<PluginDiagnosticMessage>,
) -> Result<PluginEnvDiagnostic, AppError> {
let name = trimmed_non_empty("environment variable name", &requirement.name)?;
let value = reader.get(&name);
let present = value.is_some();
let matches_expected = match (&value, &requirement.equals) {
(Some(value), Some(expected)) => value == expected,
(Some(_), None) => true,
(None, _) => false,
};
let ok = matches_expected;
let status = if ok {
"ok"
} else if present {
"mismatch"
} else {
"missing"
}
.to_owned();
if requirement.required && !ok {
messages.push(PluginDiagnosticMessage {
level: "error".to_owned(),
message: format!("environment variable {name} is {status}"),
});
}
Ok(PluginEnvDiagnostic {
name,
present,
ok,
required: requirement.required,
value,
status,
})
}
fn trimmed_non_empty(label: &str, value: &str) -> Result<String, AppError> {
let trimmed = value.trim();
if trimmed.is_empty() {
return Err(AppError::Invalid(format!("{label} must not be empty")));
}
Ok(trimmed.to_owned())
}
fn normalize_expected_kind(kind: Option<String>) -> Result<Option<String>, AppError> {
kind.map(|kind| {
let kind = kind.trim().to_ascii_lowercase();
match kind.as_str() {
"file" | "directory" | "any" => Ok(kind),
_ => Err(AppError::Invalid(format!(
"file prerequisite kind must be file, directory, or any: {kind}"
))),
}
})
.transpose()
}
fn metadata_kind(metadata: &FileMetadata) -> &'static str {
if metadata.is_file {
"file"
} else if metadata.is_dir {
"directory"
} else {
"other"
}
}
fn bounded_utf8(bytes: Vec<u8>) -> Option<String> {
let text = String::from_utf8_lossy(&bytes).trim().to_owned();
if text.is_empty() {
return None;
}
const LIMIT: usize = 4096;
if text.len() <= LIMIT {
Some(text)
} else {
let mut end = LIMIT;
while !text.is_char_boundary(end) {
end -= 1;
}
Some(text[..end].to_owned())
}
}
fn first_non_empty_line(text: Option<&str>) -> Option<String> {
text.and_then(|text| {
text.lines()
.map(str::trim)
.find(|line| !line.is_empty())
.map(str::to_owned)
})
}
/// Input for subscribing to public plugin events.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginEventSubscribeInput {
/// Project id whose public events should be observed.
pub project_id: String,
/// Event types to keep. Empty means every supported public plugin event.
#[serde(default)]
pub event_types: Vec<String>,
/// Per-subscription retained event capacity. Defaults to 100, capped at 1000.
#[serde(default)]
pub capacity: Option<usize>,
}
/// Input for polling one public plugin event subscription.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginEventPollInput {
/// Subscription id returned by subscribe.
pub subscription_id: String,
/// Maximum number of events to drain. Defaults to 100, capped at 1000.
#[serde(default)]
pub max_events: Option<usize>,
}
/// Input for unsubscribing from public plugin events.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginEventUnsubscribeInput {
/// Subscription id returned by subscribe.
pub subscription_id: String,
}
/// Active public plugin event subscription.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginEventSubscription {
/// Opaque subscription id.
pub subscription_id: String,
/// Project id observed by this subscription.
pub project_id: String,
/// Event types retained by this subscription.
pub event_types: Vec<String>,
/// Per-subscription retained event capacity.
pub capacity: usize,
/// Delivery guarantee label.
pub retention: String,
}
/// Batch drained from one public plugin event subscription.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginEventBatch {
/// Subscription id.
pub subscription_id: String,
/// Drained events, oldest first.
pub events: Vec<PluginPublicEvent>,
/// Number of older retained events dropped since the previous poll.
pub dropped: usize,
}
/// Public event visible to plugins.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum PluginPublicEvent {
/// A workspace file changed through the public plugin workspace API.
#[serde(rename_all = "camelCase")]
WorkspaceFileChanged {
/// Monotonic sequence allocated by the public plugin event facade.
sequence: u64,
/// Event observation time, epoch milliseconds.
occurred_at_ms: i64,
/// Project id.
project_id: String,
/// Normalized relative workspace path.
path: String,
/// Public operation label.
operation: String,
},
/// A background task lifecycle event occurred.
#[serde(rename_all = "camelCase")]
BackgroundTaskChanged {
/// Monotonic sequence allocated by the public plugin event facade.
sequence: u64,
/// Event observation time, epoch milliseconds.
occurred_at_ms: i64,
/// Project id.
project_id: String,
/// Task id.
task_id: String,
/// Owner agent id.
owner_agent_id: String,
/// Public task state/event label.
state: String,
},
}
impl PluginPublicEvent {
fn event_type(&self) -> &'static str {
match self {
Self::WorkspaceFileChanged { .. } => "workspaceFileChanged",
Self::BackgroundTaskChanged { .. } => "backgroundTaskChanged",
}
}
fn project_id(&self) -> &str {
match self {
Self::WorkspaceFileChanged { project_id, .. }
| Self::BackgroundTaskChanged { project_id, .. } => project_id,
}
}
}
struct PluginEventSubscriptionState {
project_id: String,
event_types: HashSet<String>,
capacity: usize,
queue: VecDeque<PluginPublicEvent>,
dropped: usize,
}
/// Public plugin event subscription facade.
pub struct PluginEventSubscriptions {
projects: Arc<dyn ProjectStore>,
ids: Arc<dyn IdGenerator>,
clock: Arc<dyn Clock>,
subscriptions: Mutex<HashMap<String, PluginEventSubscriptionState>>,
sequence: Mutex<u64>,
}
impl PluginEventSubscriptions {
/// Builds the facade.
#[must_use]
pub fn new(
projects: Arc<dyn ProjectStore>,
ids: Arc<dyn IdGenerator>,
clock: Arc<dyn Clock>,
) -> Self {
Self {
projects,
ids,
clock,
subscriptions: Mutex::new(HashMap::new()),
sequence: Mutex::new(0),
}
}
/// Creates a disposable public event subscription.
///
/// # Errors
/// [`AppError`] for malformed projects, unknown projects, or unsupported event types.
pub async fn subscribe(
&self,
input: PluginEventSubscribeInput,
) -> Result<PluginEventSubscription, AppError> {
let project_id = parse_project_id(&input.project_id)?;
self.projects.load_project(project_id).await?;
let event_types = normalize_event_types(input.event_types)?;
let capacity = input.capacity.unwrap_or(100).clamp(1, 1000);
let subscription_id = self.ids.new_uuid().to_string();
self.subscriptions.lock().unwrap().insert(
subscription_id.clone(),
PluginEventSubscriptionState {
project_id: input.project_id.clone(),
event_types: event_types.iter().cloned().collect(),
capacity,
queue: VecDeque::new(),
dropped: 0,
},
);
Ok(PluginEventSubscription {
subscription_id,
project_id: input.project_id,
event_types: if event_types.is_empty() {
supported_plugin_event_types()
} else {
event_types
},
capacity,
retention: "bestEffortBounded".to_owned(),
})
}
/// Drains retained events for one subscription.
///
/// # Errors
/// [`AppError::NotFound`] when the subscription does not exist.
pub fn poll(&self, input: PluginEventPollInput) -> Result<PluginEventBatch, AppError> {
let max_events = input.max_events.unwrap_or(100).clamp(1, 1000);
let mut subscriptions = self.subscriptions.lock().unwrap();
let subscription = subscriptions
.get_mut(&input.subscription_id)
.ok_or_else(|| AppError::NotFound("plugin event subscription".to_owned()))?;
let take = max_events.min(subscription.queue.len());
let events = subscription.queue.drain(..take).collect();
let dropped = std::mem::take(&mut subscription.dropped);
Ok(PluginEventBatch {
subscription_id: input.subscription_id,
events,
dropped,
})
}
/// Disposes one subscription. Unknown subscriptions are treated as already disposed.
pub fn unsubscribe(&self, input: PluginEventUnsubscribeInput) -> PluginEventSubscription {
let existed = self
.subscriptions
.lock()
.unwrap()
.remove(&input.subscription_id)
.map(|state| PluginEventSubscription {
subscription_id: input.subscription_id.clone(),
project_id: state.project_id,
event_types: if state.event_types.is_empty() {
supported_plugin_event_types()
} else {
sorted_event_types(state.event_types)
},
capacity: state.capacity,
retention: "disposed".to_owned(),
});
existed.unwrap_or(PluginEventSubscription {
subscription_id: input.subscription_id,
project_id: String::new(),
event_types: Vec::new(),
capacity: 0,
retention: "disposed".to_owned(),
})
}
/// Records a domain event after projecting it to the stable public plugin contract.
pub fn record_domain_event(&self, event: &DomainEvent) {
let Some(public) = self.public_event_from_domain(event) else {
return;
};
let mut subscriptions = self.subscriptions.lock().unwrap();
for subscription in subscriptions.values_mut() {
if subscription.accepts(&public) {
if subscription.queue.len() >= subscription.capacity {
subscription.queue.pop_front();
subscription.dropped += 1;
}
subscription.queue.push_back(public.clone());
}
}
}
fn public_event_from_domain(&self, event: &DomainEvent) -> Option<PluginPublicEvent> {
if !is_supported_domain_event(event) {
return None;
}
let mut sequence = self.sequence.lock().unwrap();
*sequence += 1;
let sequence = *sequence;
let occurred_at_ms = self.clock.now_millis();
match event {
DomainEvent::PluginWorkspaceFileChanged {
project_id,
path,
operation,
} => Some(PluginPublicEvent::WorkspaceFileChanged {
sequence,
occurred_at_ms,
project_id: project_id.to_string(),
path: path.clone(),
operation: operation.clone(),
}),
DomainEvent::BackgroundTaskStarted {
project_id,
task_id,
owner_agent_id,
} => Some(PluginPublicEvent::BackgroundTaskChanged {
sequence,
occurred_at_ms,
project_id: project_id.to_string(),
task_id: task_id.to_string(),
owner_agent_id: owner_agent_id.to_string(),
state: "started".to_owned(),
}),
DomainEvent::BackgroundTaskStateChanged {
project_id,
task_id,
owner_agent_id,
state,
} => Some(PluginPublicEvent::BackgroundTaskChanged {
sequence,
occurred_at_ms,
project_id: project_id.to_string(),
task_id: task_id.to_string(),
owner_agent_id: owner_agent_id.to_string(),
state: background_task_state_label(*state).to_owned(),
}),
DomainEvent::BackgroundTaskCompleted {
project_id,
task_id,
owner_agent_id,
..
} => Some(PluginPublicEvent::BackgroundTaskChanged {
sequence,
occurred_at_ms,
project_id: project_id.to_string(),
task_id: task_id.to_string(),
owner_agent_id: owner_agent_id.to_string(),
state: "completed".to_owned(),
}),
DomainEvent::BackgroundTaskFailed {
project_id,
task_id,
owner_agent_id,
..
} => Some(PluginPublicEvent::BackgroundTaskChanged {
sequence,
occurred_at_ms,
project_id: project_id.to_string(),
task_id: task_id.to_string(),
owner_agent_id: owner_agent_id.to_string(),
state: "failed".to_owned(),
}),
DomainEvent::BackgroundTaskCancelled {
project_id,
task_id,
owner_agent_id,
..
} => Some(PluginPublicEvent::BackgroundTaskChanged {
sequence,
occurred_at_ms,
project_id: project_id.to_string(),
task_id: task_id.to_string(),
owner_agent_id: owner_agent_id.to_string(),
state: "cancelled".to_owned(),
}),
DomainEvent::BackgroundTaskCompletionDeliveryPending {
project_id,
task_id,
owner_agent_id,
} => Some(PluginPublicEvent::BackgroundTaskChanged {
sequence,
occurred_at_ms,
project_id: project_id.to_string(),
task_id: task_id.to_string(),
owner_agent_id: owner_agent_id.to_string(),
state: "deliveryPending".to_owned(),
}),
DomainEvent::BackgroundTaskCompletionDelivered {
project_id,
task_id,
owner_agent_id,
} => Some(PluginPublicEvent::BackgroundTaskChanged {
sequence,
occurred_at_ms,
project_id: project_id.to_string(),
task_id: task_id.to_string(),
owner_agent_id: owner_agent_id.to_string(),
state: "delivered".to_owned(),
}),
_ => None,
}
}
}
impl PluginEventSubscriptionState {
fn accepts(&self, event: &PluginPublicEvent) -> bool {
self.project_id == event.project_id()
&& (self.event_types.is_empty() || self.event_types.contains(event.event_type()))
}
}
fn normalize_event_types(event_types: Vec<String>) -> Result<Vec<String>, AppError> {
let supported: HashSet<String> = supported_plugin_event_types().into_iter().collect();
let mut normalized = Vec::new();
for event_type in event_types {
let event_type = event_type.trim().to_owned();
if event_type.is_empty() {
continue;
}
if !supported.contains(&event_type) {
return Err(AppError::Invalid(format!(
"unsupported plugin event type: {event_type}"
)));
}
if !normalized.contains(&event_type) {
normalized.push(event_type);
}
}
Ok(normalized)
}
fn supported_plugin_event_types() -> Vec<String> {
vec![
"workspaceFileChanged".to_owned(),
"backgroundTaskChanged".to_owned(),
]
}
fn is_supported_domain_event(event: &DomainEvent) -> bool {
matches!(
event,
DomainEvent::PluginWorkspaceFileChanged { .. }
| DomainEvent::BackgroundTaskStarted { .. }
| DomainEvent::BackgroundTaskStateChanged { .. }
| DomainEvent::BackgroundTaskCompleted { .. }
| DomainEvent::BackgroundTaskFailed { .. }
| DomainEvent::BackgroundTaskCancelled { .. }
| DomainEvent::BackgroundTaskCompletionDeliveryPending { .. }
| DomainEvent::BackgroundTaskCompletionDelivered { .. }
)
}
fn sorted_event_types(types: HashSet<String>) -> Vec<String> {
let mut types: Vec<_> = types.into_iter().collect();
types.sort();
types
}
fn background_task_state_label(state: BackgroundTaskState) -> &'static str {
match state {
BackgroundTaskState::Queued => "queued",
BackgroundTaskState::Running => "running",
BackgroundTaskState::Waiting => "waiting",
BackgroundTaskState::Completed => "completed",
BackgroundTaskState::Failed => "failed",
BackgroundTaskState::Cancelled => "cancelled",
BackgroundTaskState::Expired => "expired",
}
}
/// Public plugin facade for command-backed background tasks.
pub struct PluginCommandTasks {
projects: Arc<dyn ProjectStore>,
tasks: Arc<dyn BackgroundTaskStore>,
spawn: Arc<SpawnBackgroundCommand>,
}
impl PluginCommandTasks {
/// Builds the facade.
#[must_use]
pub fn new(
projects: Arc<dyn ProjectStore>,
tasks: Arc<dyn BackgroundTaskStore>,
spawn: Arc<SpawnBackgroundCommand>,
) -> Self {
Self {
projects,
tasks,
spawn,
}
}
/// Launches a command as a first-class background task.
///
/// # Errors
/// [`AppError`] for malformed ids, unsafe paths, unknown projects, or runner failures.
pub async fn run_command(
&self,
input: PluginRunCommandInput,
) -> Result<BackgroundTask, AppError> {
if input.command.trim().is_empty() {
return Err(AppError::Invalid("command must not be empty".to_owned()));
}
let project_id = parse_project_id(&input.project_id)?;
let owner_agent_id = parse_agent_id(&input.owner_agent_id)?;
let project = self.projects.load_project(project_id).await?;
let cwd_relative = input.cwd.as_deref().unwrap_or(".");
let cwd = resolve_workspace_path(&project, cwd_relative)?;
let cwd = ProjectPath::new(cwd.absolute.as_str().to_owned())
.map_err(|err| AppError::Invalid(err.to_string()))?;
let command = SpawnSpec {
command: input.command,
args: input.args,
cwd,
env: input.env,
context_plan: None,
sandbox: None,
};
let wake_policy = if input.record_only {
BackgroundTaskWakePolicy::RecordOnly
} else {
BackgroundTaskWakePolicy::WakeOwner
};
self.spawn
.execute(SpawnBackgroundCommandInput {
project_id,
owner_agent_id,
label: input.label,
command,
wake_policy,
rendezvous: None,
deadline_ms: input.deadline_ms,
})
.await
.map(|out| out.task)
}
/// Reads one task status by id.
///
/// # Errors
/// [`AppError`] for malformed task ids or store failures.
pub async fn get_status(
&self,
input: PluginTaskStatusInput,
) -> Result<Option<BackgroundTask>, AppError> {
let task_id = parse_task_id(&input.task_id)?;
self.tasks
.get(task_id)
.await
.map_err(map_background_task_err)
}
}
fn parse_agent_id(raw: &str) -> Result<AgentId, AppError> {
Uuid::parse_str(raw)
.map(AgentId::from_uuid)
.map_err(|_| AppError::Invalid(format!("invalid agent id: {raw}")))
}
fn parse_task_id(raw: &str) -> Result<TaskId, AppError> {
Uuid::parse_str(raw)
.map(TaskId::from_uuid)
.map_err(|_| AppError::Invalid(format!("invalid task id: {raw}")))
}
fn map_background_task_err(err: domain::ports::BackgroundTaskPortError) -> AppError {
match err {
domain::ports::BackgroundTaskPortError::NotFound => {
AppError::NotFound("background task".to_owned())
}
domain::ports::BackgroundTaskPortError::AlreadyExists => {
AppError::Invalid("background task already exists".to_owned())
}
domain::ports::BackgroundTaskPortError::Invalid(message) => AppError::Invalid(message),
domain::ports::BackgroundTaskPortError::Runner(message) => AppError::Process(message),
domain::ports::BackgroundTaskPortError::Store(message) => AppError::Store(message),
}
}
fn map_store(e: PluginStoreError) -> AppError {
match e {
PluginStoreError::NotFound => AppError::NotFound("plugin package".to_owned()),
PluginStoreError::Invalid(m) | PluginStoreError::Format(m) => AppError::Invalid(m),
PluginStoreError::Io(m) => AppError::FileSystem(m),
}
}
fn map_storage(e: PluginStorageError) -> AppError {
match e {
PluginStorageError::Invalid(m) => AppError::Invalid(m),
PluginStorageError::Io(m) => AppError::FileSystem(m),
PluginStorageError::Serialization(m) => AppError::Store(m),
}
}
fn map_registry(e: PluginRegistryError) -> AppError {
match e {
PluginRegistryError::Io(m) => AppError::Store(m),
PluginRegistryError::Serialization(m) => AppError::Store(m),
}
}
fn map_manifest(e: PluginManifestError) -> AppError {
match e {
PluginManifestError::Json(m)
| PluginManifestError::Invalid(m)
| PluginManifestError::IncompatibleEngine(m) => AppError::Invalid(m),
}
}
fn map_mcp(e: PluginMcpError) -> AppError {
AppError::Process(e.to_string())
}
fn plugin_package_ref(id: &PluginId) -> domain::PluginPackageRef {
domain::PluginPackageRef {
plugin_id: Some(id.clone()),
root: id.as_str().to_owned(),
}
}
fn admin_from_descriptor(
d: PluginDescriptor,
_packages: &dyn PluginPackageStore,
) -> Result<PluginAdmin, AppError> {
let icon_url = match &d.manifest.icon {
Some(icon) => Some(plugin_asset_url(
&d.manifest.id,
d.manifest.version.as_str(),
&d.registry.content_hash,
icon,
)),
None => None,
};
Ok(PluginAdmin {
id: d.manifest.id.as_str().to_owned(),
display_name: d.manifest.display_name,
publisher: d.manifest.publisher,
version: d.manifest.version.as_str().to_owned(),
description: d.manifest.description,
icon_url,
source_kind: d.registry.source.kind().to_owned(),
source_label: Some(d.registry.source.label().to_owned()),
lifecycle_state: d.registry.lifecycle_state,
enabled: matches!(
d.registry.lifecycle_state,
PluginLifecycleState::Enabled | PluginLifecycleState::PendingEnable
),
pending_enable_state: match d.registry.lifecycle_state {
PluginLifecycleState::PendingEnable => Some(true),
PluginLifecycleState::PendingDisable => Some(false),
_ => None,
},
pending_uninstall: d.registry.lifecycle_state == PluginLifecycleState::PendingUninstall,
restart_required: d.registry.restart_required,
trust_level: d.manifest.trust_level,
contribution_summary: PluginContributionSummary::from(&d.manifest.contributes),
error: d.registry.error,
})
}
fn plugin_asset_url(
plugin_id: &PluginId,
version: &str,
hash: &ContentHash,
path: &domain::RelativePath,
) -> String {
format!(
"idea-plugin://{}/{}/{}/{}",
plugin_id.as_str(),
version,
hash.as_str(),
path.as_str()
)
}
async fn descriptor_for(
packages: &dyn PluginPackageStore,
validator: &dyn PluginManifestValidator,
entry: PluginRegistryEntry,
) -> Result<PluginDescriptor, AppError> {
let bytes = packages
.read_manifest(&plugin_package_ref(&entry.id))
.await
.map_err(map_store)?;
let manifest = validator
.validate(&bytes.bytes, &plugin_package_ref(&entry.id))
.map_err(map_manifest)?;
Ok(PluginDescriptor {
manifest,
registry: entry,
})
}
/// Lists admin plugins.
pub struct ListPlugins {
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
}
impl ListPlugins {
/// Builds the use case.
#[must_use]
pub fn new(
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
) -> Self {
Self {
packages,
registry,
validator,
}
}
/// Executes the use case.
pub async fn execute(&self) -> Result<Vec<PluginAdmin>, AppError> {
let registry = self.registry.load_registry().await.map_err(map_registry)?;
let mut out = Vec::new();
for entry in registry.plugins {
match descriptor_for(
self.packages.as_ref(),
self.validator.as_ref(),
entry.clone(),
)
.await
{
Ok(d) => out.push(admin_from_descriptor(d, self.packages.as_ref())?),
Err(err) => {
let invalid = PluginRegistryEntry {
lifecycle_state: PluginLifecycleState::Invalid,
error: Some(err.to_string()),
..entry
};
let placeholder = PluginManifest {
idea_plugin_manifest_version: 1,
id: invalid.id.clone(),
display_name: invalid.id.as_str().to_owned(),
publisher: None,
version: domain::PluginVersion::new("0.0.0")
.expect("literal semver is valid"),
description: None,
engine_idea: None,
main: domain::RelativePath::new("dist/index.js")
.expect("literal path is valid"),
icon: None,
trust_level: PluginTrustLevel::Full,
capabilities: Vec::new(),
activation_scope: domain::PluginActivationScope::default(),
contributes: PluginContributionSet::default(),
};
out.push(admin_from_descriptor(
PluginDescriptor {
manifest: placeholder,
registry: invalid,
},
self.packages.as_ref(),
)?);
}
}
}
out.sort_by(|a, b| a.display_name.cmp(&b.display_name).then(a.id.cmp(&b.id)));
Ok(out)
}
}
/// Reviews a plugin package without committing it.
pub struct ReviewPluginPackage {
packages: Arc<dyn PluginPackageStore>,
validator: Arc<dyn PluginManifestValidator>,
}
impl ReviewPluginPackage {
/// Builds the use case.
#[must_use]
pub fn new(
packages: Arc<dyn PluginPackageStore>,
validator: Arc<dyn PluginManifestValidator>,
) -> Self {
Self {
packages,
validator,
}
}
/// Executes the use case.
pub async fn execute(&self, input: ReviewPluginPackageInput) -> Result<PluginReview, AppError> {
let staged = match input {
ReviewPluginPackageInput::Archive { path } => self
.packages
.install_from_archive(&LocalPath::new(path))
.await
.map_err(map_store)?,
ReviewPluginPackageInput::Directory { path } => self
.packages
.install_from_directory(&LocalPath::new(path))
.await
.map_err(map_store)?,
};
review_staged(self.packages.as_ref(), self.validator.as_ref(), &staged).await
}
}
async fn review_staged(
packages: &dyn PluginPackageStore,
validator: &dyn PluginManifestValidator,
staged: &StagedPluginPackage,
) -> Result<PluginReview, AppError> {
let package = domain::PluginPackageRef {
plugin_id: None,
root: staged.root.clone(),
};
let PluginManifestBytes { bytes } =
packages.read_manifest(&package).await.map_err(map_store)?;
let manifest = validator.validate(&bytes, &package).map_err(map_manifest)?;
Ok(PluginReview {
contribution_summary: PluginContributionSummary::from(&manifest.contributes),
trust_level: manifest.trust_level,
manifest,
source: staged.source.clone(),
content_hash: staged.content_hash.as_str().to_owned(),
})
}
/// Installs from archive.
pub struct InstallPluginFromArchive {
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
events: Arc<dyn EventBus>,
mcp: Arc<dyn PluginMcpSupervisor>,
}
impl InstallPluginFromArchive {
/// Builds the use case.
#[must_use]
pub fn new(
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
events: Arc<dyn EventBus>,
mcp: Arc<dyn PluginMcpSupervisor>,
) -> Self {
Self {
packages,
registry,
validator,
events,
mcp,
}
}
/// Executes the use case.
pub async fn execute(&self, path: String) -> Result<PluginInstallResult, AppError> {
install_from_staged(
self.packages.as_ref(),
self.registry.as_ref(),
self.validator.as_ref(),
self.events.as_ref(),
self.mcp.as_ref(),
self.packages
.install_from_archive(&LocalPath::new(path))
.await
.map_err(map_store)?,
)
.await
}
}
/// Installs from directory.
pub struct InstallPluginFromDirectory {
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
events: Arc<dyn EventBus>,
mcp: Arc<dyn PluginMcpSupervisor>,
}
impl InstallPluginFromDirectory {
/// Builds the use case.
#[must_use]
pub fn new(
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
events: Arc<dyn EventBus>,
mcp: Arc<dyn PluginMcpSupervisor>,
) -> Self {
Self {
packages,
registry,
validator,
events,
mcp,
}
}
/// Executes the use case.
pub async fn execute(&self, path: String) -> Result<PluginInstallResult, AppError> {
install_from_staged(
self.packages.as_ref(),
self.registry.as_ref(),
self.validator.as_ref(),
self.events.as_ref(),
self.mcp.as_ref(),
self.packages
.install_from_directory(&LocalPath::new(path))
.await
.map_err(map_store)?,
)
.await
}
}
async fn install_from_staged(
packages: &dyn PluginPackageStore,
registry_store: &dyn PluginRegistryStore,
validator: &dyn PluginManifestValidator,
events: &dyn EventBus,
mcp: &dyn PluginMcpSupervisor,
staged: StagedPluginPackage,
) -> Result<PluginInstallResult, AppError> {
let review = review_staged(packages, validator, &staged).await?;
let plugin_id = review.manifest.id.clone();
crate::diag!(
"[plugins] install staged reviewed plugin={} version={} hash={} source={}",
plugin_id.as_str(),
review.manifest.version.as_str(),
review.content_hash,
review.source.kind()
);
packages
.commit_install(staged, &plugin_id)
.await
.map_err(map_store)?;
crate::diag!("[plugins] install committed plugin={}", plugin_id.as_str());
let mut registry = registry_store.load_registry().await.map_err(map_registry)?;
let mut entry = PluginRegistryEntry {
id: plugin_id.clone(),
lifecycle_state: PluginLifecycleState::Enabled,
source: review.source.clone(),
content_hash: ContentHash::new(review.content_hash.clone())
.map_err(|e| AppError::Invalid(e.to_string()))?,
restart_required: true,
error: None,
};
if let Err(err) = runtime_plugin_from_entry(packages, validator, entry.clone()).await {
let message = format!(
"runtime contributions disabled: plugin `{}` is not servable: {err}",
plugin_id.as_str()
);
crate::diag!("[plugins] {message}");
entry.lifecycle_state = PluginLifecycleState::Invalid;
entry.error = Some(message);
}
registry.upsert(entry.clone());
registry_store
.save_registry(&registry)
.await
.map_err(map_registry)?;
crate::diag!(
"[plugins] install registry saved plugin={} lifecycle={:?}",
plugin_id.as_str(),
entry.lifecycle_state
);
events.publish(DomainEvent::PluginInstalled {
plugin_id: plugin_id.clone(),
version: review.manifest.version.clone(),
});
let (active_servers, invalid_plugins) =
active_mcp_specs(packages, validator, &registry).await?;
persist_invalid_runtime_plugins(registry_store, &mut registry, invalid_plugins).await;
let active_server_count = active_servers.len();
match mcp.reconcile(active_servers).await {
Ok(statuses) => crate::diag!(
"[plugins] install MCP reconcile ok plugin={} requested={} statuses={}",
plugin_id.as_str(),
active_server_count,
statuses.servers.len()
),
Err(err) => crate::diag!(
"[plugins] install MCP reconcile failed plugin={} requested={} error={}",
plugin_id.as_str(),
active_server_count,
err
),
}
let admin = admin_from_descriptor(
PluginDescriptor {
manifest: review.manifest.clone(),
registry: entry,
},
packages,
)?;
Ok(PluginInstallResult {
plugin: admin,
review,
restart_required: true,
})
}
/// Enable/disable input.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SetPluginEnabledInput {
/// Plugin id.
pub plugin_id: String,
/// Desired enabled state.
pub enabled: bool,
}
/// Enables or disables a plugin.
pub struct SetPluginEnabled {
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
events: Arc<dyn EventBus>,
mcp: Arc<dyn PluginMcpSupervisor>,
}
impl SetPluginEnabled {
/// Builds the use case.
#[must_use]
pub fn new(
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
events: Arc<dyn EventBus>,
mcp: Arc<dyn PluginMcpSupervisor>,
) -> Self {
Self {
packages,
registry,
validator,
events,
mcp,
}
}
/// Executes the use case.
pub async fn execute(&self, input: SetPluginEnabledInput) -> Result<PluginAdmin, AppError> {
let plugin_id =
PluginId::new(input.plugin_id).map_err(|e| AppError::Invalid(e.to_string()))?;
let mut registry = self.registry.load_registry().await.map_err(map_registry)?;
let entry = registry
.plugins
.iter_mut()
.find(|p| p.id == plugin_id)
.ok_or_else(|| AppError::NotFound("plugin".to_owned()))?;
entry.lifecycle_state = if input.enabled {
PluginLifecycleState::Enabled
} else {
PluginLifecycleState::Disabled
};
entry.restart_required = true;
let saved = entry.clone();
self.registry
.save_registry(&registry)
.await
.map_err(map_registry)?;
if input.enabled {
self.events.publish(DomainEvent::PluginEnabled {
plugin_id: plugin_id.clone(),
});
} else {
self.mcp.stop_plugin(&plugin_id).await.map_err(map_mcp)?;
self.events.publish(DomainEvent::PluginDisabled {
plugin_id: plugin_id.clone(),
restart_required: true,
});
}
let (active_servers, invalid_plugins) =
active_mcp_specs(self.packages.as_ref(), self.validator.as_ref(), &registry).await?;
persist_invalid_runtime_plugins(self.registry.as_ref(), &mut registry, invalid_plugins)
.await;
let _ = self.mcp.reconcile(active_servers).await;
let descriptor =
descriptor_for(self.packages.as_ref(), self.validator.as_ref(), saved).await?;
admin_from_descriptor(descriptor, self.packages.as_ref())
}
}
/// Uninstall input.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UninstallPluginInput {
/// Plugin id.
pub plugin_id: String,
}
/// Plugin-owned key/value storage facade.
pub struct PluginStorageAccess {
storage: Arc<dyn PluginStorageStore>,
registry: Arc<dyn PluginRegistryStore>,
}
impl PluginStorageAccess {
/// Builds the facade.
#[must_use]
pub fn new(
storage: Arc<dyn PluginStorageStore>,
registry: Arc<dyn PluginRegistryStore>,
) -> Self {
Self { storage, registry }
}
/// Reads one plugin-owned JSON value.
pub async fn get(
&self,
input: PluginStorageGetInput,
) -> Result<Option<serde_json::Value>, AppError> {
let plugin_id = self.active_plugin_id(input.plugin_id).await?;
validate_storage_key(&input.key)?;
self.storage
.get(&plugin_id, &input.key)
.await
.map_err(map_storage)
}
/// Writes one plugin-owned JSON value.
pub async fn set(&self, input: PluginStorageSetInput) -> Result<(), AppError> {
let plugin_id = self.active_plugin_id(input.plugin_id).await?;
validate_storage_key(&input.key)?;
self.storage
.set(&plugin_id, &input.key, input.value)
.await
.map_err(map_storage)
}
/// Deletes one plugin-owned JSON value.
pub async fn delete(&self, input: PluginStorageGetInput) -> Result<bool, AppError> {
let plugin_id = self.active_plugin_id(input.plugin_id).await?;
validate_storage_key(&input.key)?;
self.storage
.delete(&plugin_id, &input.key)
.await
.map_err(map_storage)
}
async fn active_plugin_id(&self, raw: String) -> Result<PluginId, AppError> {
let plugin_id = PluginId::new(raw).map_err(|e| AppError::Invalid(e.to_string()))?;
let registry = self.registry.load_registry().await.map_err(map_registry)?;
let entry = registry
.find(&plugin_id)
.ok_or_else(|| AppError::NotFound("plugin".to_owned()))?;
if !entry.lifecycle_state.is_runtime_active() {
return Err(AppError::Invalid("plugin is not runtime-active".to_owned()));
}
Ok(plugin_id)
}
}
fn validate_storage_key(key: &str) -> Result<(), AppError> {
if key.trim().is_empty() {
return Err(AppError::Invalid(
"plugin storage key must not be empty".to_owned(),
));
}
if key.len() > 512 {
return Err(AppError::Invalid(
"plugin storage key must not exceed 512 bytes".to_owned(),
));
}
if key.contains('\0') {
return Err(AppError::Invalid(
"plugin storage key must not contain NUL bytes".to_owned(),
));
}
Ok(())
}
/// Uninstalls a plugin.
pub struct UninstallPlugin {
packages: Arc<dyn PluginPackageStore>,
storage: Arc<dyn PluginStorageStore>,
registry: Arc<dyn PluginRegistryStore>,
events: Arc<dyn EventBus>,
mcp: Arc<dyn PluginMcpSupervisor>,
}
impl UninstallPlugin {
/// Builds the use case.
#[must_use]
pub fn new(
packages: Arc<dyn PluginPackageStore>,
storage: Arc<dyn PluginStorageStore>,
registry: Arc<dyn PluginRegistryStore>,
events: Arc<dyn EventBus>,
mcp: Arc<dyn PluginMcpSupervisor>,
) -> Self {
Self {
packages,
storage,
registry,
events,
mcp,
}
}
/// Executes the use case.
pub async fn execute(
&self,
input: UninstallPluginInput,
) -> Result<UninstallPluginResult, AppError> {
let plugin_id =
PluginId::new(input.plugin_id).map_err(|e| AppError::Invalid(e.to_string()))?;
self.mcp.stop_plugin(&plugin_id).await.map_err(map_mcp)?;
let mut registry = self.registry.load_registry().await.map_err(map_registry)?;
registry
.remove(&plugin_id)
.ok_or_else(|| AppError::NotFound("plugin".to_owned()))?;
self.registry
.save_registry(&registry)
.await
.map_err(map_registry)?;
let removal = self
.packages
.remove_package(&plugin_id)
.await
.map_err(map_store)?;
self.storage
.purge_plugin(&plugin_id)
.await
.map_err(map_storage)?;
self.events.publish(DomainEvent::PluginUninstalled {
plugin_id: plugin_id.clone(),
restart_required: true,
});
Ok(UninstallPluginResult {
plugin_id: plugin_id.as_str().to_owned(),
removal_outcome: removal,
restart_required: true,
})
}
}
/// Lists runtime contributions.
pub struct ListPluginRuntimeContributions {
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
}
impl ListPluginRuntimeContributions {
/// Builds the use case.
#[must_use]
pub fn new(
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
) -> Self {
Self {
packages,
registry,
validator,
}
}
/// Executes the use case.
pub async fn execute(&self) -> Result<PluginRuntimeCatalog, AppError> {
let mut registry = self.registry.load_registry().await.map_err(map_registry)?;
let mut plugins = Vec::new();
let mut invalid_plugins = Vec::new();
for entry in registry.plugins.clone() {
if !entry.lifecycle_state.is_runtime_active() {
continue;
}
match runtime_plugin_from_entry(
self.packages.as_ref(),
self.validator.as_ref(),
entry.clone(),
)
.await
{
Ok(plugin) => plugins.push(plugin),
Err(err) => {
let message = format!(
"runtime contributions disabled: plugin `{}` is not servable: {err}",
entry.id.as_str()
);
crate::diag!("[plugins] {message}");
invalid_plugins.push((entry.id, message));
}
}
}
if !invalid_plugins.is_empty() {
for (plugin_id, message) in invalid_plugins {
if let Some(entry) = registry.plugins.iter_mut().find(|p| p.id == plugin_id) {
entry.lifecycle_state = PluginLifecycleState::Invalid;
entry.error = Some(message);
}
}
if let Err(err) = self
.registry
.save_registry(&registry)
.await
.map_err(map_registry)
{
crate::diag!("[plugins] failed to persist invalid runtime plugin state: {err}");
}
}
Ok(PluginRuntimeCatalog { plugins })
}
}
async fn runtime_plugin_from_entry(
packages: &dyn PluginPackageStore,
validator: &dyn PluginManifestValidator,
entry: PluginRegistryEntry,
) -> Result<PluginRuntimePlugin, AppError> {
let descriptor = descriptor_for(packages, validator, entry).await?;
let bundle = checked_plugin_asset_url(
packages,
&descriptor.manifest.id,
descriptor.manifest.version.as_str(),
&descriptor.registry.content_hash,
&descriptor.manifest.main,
)?;
let icon_url = match &descriptor.manifest.icon {
Some(icon) => Some(checked_plugin_asset_url(
packages,
&descriptor.manifest.id,
descriptor.manifest.version.as_str(),
&descriptor.registry.content_hash,
icon,
)?),
None => None,
};
Ok(PluginRuntimePlugin {
id: descriptor.manifest.id.as_str().to_owned(),
display_name: descriptor.manifest.display_name,
publisher: descriptor.manifest.publisher,
version: descriptor.manifest.version.as_str().to_owned(),
bundle_url: bundle,
icon_url,
content_hash: descriptor.registry.content_hash.as_str().to_owned(),
capabilities: descriptor.manifest.capabilities,
activation_scope: descriptor.manifest.activation_scope,
contributes: descriptor.manifest.contributes,
})
}
fn checked_plugin_asset_url(
packages: &dyn PluginPackageStore,
plugin_id: &PluginId,
version: &str,
hash: &ContentHash,
path: &domain::RelativePath,
) -> Result<String, AppError> {
packages
.bundle_url(plugin_id, path, hash)
.map_err(map_store)?;
Ok(plugin_asset_url(plugin_id, version, hash, path))
}
/// Reconciles plugin MCP servers.
pub struct ReconcilePluginMcpServers {
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
mcp: Arc<dyn PluginMcpSupervisor>,
}
impl ReconcilePluginMcpServers {
/// Builds the use case.
#[must_use]
pub fn new(
packages: Arc<dyn PluginPackageStore>,
registry: Arc<dyn PluginRegistryStore>,
validator: Arc<dyn PluginManifestValidator>,
mcp: Arc<dyn PluginMcpSupervisor>,
) -> Self {
Self {
packages,
registry,
validator,
mcp,
}
}
/// Executes the use case.
pub async fn execute(&self) -> Result<domain::PluginMcpStatusSet, AppError> {
let mut registry = self.registry.load_registry().await.map_err(map_registry)?;
let (specs, invalid_plugins) =
active_mcp_specs(self.packages.as_ref(), self.validator.as_ref(), &registry).await?;
persist_invalid_runtime_plugins(self.registry.as_ref(), &mut registry, invalid_plugins)
.await;
let requested = specs.len();
let result = self.mcp.reconcile(specs).await.map_err(map_mcp);
match &result {
Ok(statuses) => crate::diag!(
"[plugins] MCP reconcile ok requested={} statuses={}",
requested,
statuses.servers.len()
),
Err(err) => {
crate::diag!("[plugins] MCP reconcile failed requested={requested} error={err}")
}
}
result
}
}
async fn active_mcp_specs(
packages: &dyn PluginPackageStore,
validator: &dyn PluginManifestValidator,
registry: &domain::PluginRegistry,
) -> Result<(Vec<PluginMcpServerSpec>, Vec<(PluginId, String)>), AppError> {
let installed_roots = packages
.list_installed()
.await
.map_err(map_store)?
.into_iter()
.filter_map(|p| p.plugin_id.clone().map(|id| (id, p.root)))
.collect::<std::collections::HashMap<_, _>>();
let app_data_dir = packages.app_data_dir_label();
let mut specs = Vec::new();
let mut invalid_plugins = Vec::new();
for entry in &registry.plugins {
if !entry.lifecycle_state.is_runtime_active() {
continue;
}
let descriptor = match descriptor_for(packages, validator, entry.clone()).await {
Ok(descriptor) => descriptor,
Err(err) => {
let message = format!(
"MCP servers disabled: plugin `{}` is not servable: {err}",
entry.id.as_str()
);
crate::diag!("[plugins] {message}");
invalid_plugins.push((entry.id.clone(), message));
continue;
}
};
let plugin_root = installed_roots
.get(&descriptor.manifest.id)
.cloned()
.unwrap_or_else(|| plugin_package_ref(&descriptor.manifest.id).root);
for server in descriptor.manifest.contributes.mcp_servers {
if !server.auto_start {
continue;
}
let command = substitute_vars(&server.command, &plugin_root, app_data_dir.as_deref());
let command = if server.allow_absolute_command || looks_absolute(&command) {
command
} else {
format!("{}/{}", plugin_root.trim_end_matches(['/', '\\']), command)
};
specs.push(PluginMcpServerSpec {
identity: format!(
"plugin:{}:{}",
descriptor.manifest.id.as_str(),
server.id.as_str()
),
plugin_id: descriptor.manifest.id.clone(),
server_id: server.id,
display_name: server.display_name,
command,
args: server
.args
.into_iter()
.map(|a| substitute_vars(&a, &plugin_root, app_data_dir.as_deref()))
.collect(),
env: server
.env
.into_iter()
.map(|(k, v)| {
(
k,
substitute_vars(&v, &plugin_root, app_data_dir.as_deref()),
)
})
.collect(),
cwd: substitute_vars(
server.cwd.as_deref().unwrap_or("${pluginRoot}"),
&plugin_root,
app_data_dir.as_deref(),
),
transport: server.transport,
});
}
}
Ok((specs, invalid_plugins))
}
async fn persist_invalid_runtime_plugins(
registry_store: &dyn PluginRegistryStore,
registry: &mut domain::PluginRegistry,
invalid_plugins: Vec<(PluginId, String)>,
) {
if invalid_plugins.is_empty() {
return;
}
for (plugin_id, message) in invalid_plugins {
if let Some(entry) = registry.plugins.iter_mut().find(|p| p.id == plugin_id) {
entry.lifecycle_state = PluginLifecycleState::Invalid;
entry.error = Some(message);
}
}
if let Err(err) = registry_store
.save_registry(registry)
.await
.map_err(map_registry)
{
crate::diag!("[plugins] failed to persist invalid runtime plugin state: {err}");
}
}
fn substitute_vars(raw: &str, plugin_root: &str, app_data_dir: Option<&str>) -> String {
let value = raw.replace("${pluginRoot}", plugin_root);
match app_data_dir {
Some(app_data_dir) => value.replace("${appDataDir}", app_data_dir),
None => value,
}
}
/// JSON manifest validator.
#[derive(Debug, Clone)]
pub struct JsonPluginManifestValidator {
idea_version: String,
}
impl JsonPluginManifestValidator {
/// Builds a validator for the current app version.
#[must_use]
pub fn new(idea_version: impl Into<String>) -> Self {
Self {
idea_version: idea_version.into(),
}
}
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct RawManifest {
idea_plugin_manifest_version: u32,
id: String,
display_name: String,
#[serde(default)]
publisher: Option<String>,
version: String,
#[serde(default)]
description: Option<String>,
#[serde(default)]
engines: RawEngines,
main: String,
#[serde(default)]
icon: Option<String>,
trust_level: String,
#[serde(default)]
capabilities: Vec<String>,
#[serde(default)]
activation_scope: domain::PluginActivationScope,
contributes: RawContributes,
}
#[derive(Debug, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
struct RawEngines {
#[serde(default)]
idea: Option<String>,
}
#[derive(Debug, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
struct RawContributes {
#[serde(default)]
menus: Vec<RawMenu>,
#[serde(default)]
menu_items: Vec<RawMenuItem>,
#[serde(default)]
layouts: Vec<RawLayout>,
#[serde(default)]
mcp_servers: Vec<RawMcpServer>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct RawMenu {
id: String,
label: String,
top_level: bool,
#[serde(default)]
order: Option<i32>,
#[serde(default)]
icon: Option<String>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct RawMenuItem {
id: String,
target_menu_id: String,
label: String,
command: String,
#[serde(default)]
order: Option<i32>,
#[serde(default)]
icon: Option<String>,
#[serde(default)]
when: Option<String>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct RawLayout {
#[serde(rename = "type")]
layout_type: String,
label: String,
component: String,
#[serde(default)]
order: Option<i32>,
#[serde(default)]
icon: Option<String>,
#[serde(default)]
when: Option<String>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct RawMcpServer {
id: String,
display_name: String,
command: String,
#[serde(default)]
args: Vec<String>,
#[serde(default)]
env: std::collections::BTreeMap<String, String>,
#[serde(default)]
cwd: Option<String>,
transport: String,
#[serde(default)]
auto_start: bool,
#[serde(default)]
allow_absolute_command: bool,
}
impl PluginManifestValidator for JsonPluginManifestValidator {
fn validate(
&self,
bytes: &[u8],
_package_root: &domain::PluginPackageRef,
) -> Result<PluginManifest, PluginManifestError> {
let raw: RawManifest =
serde_json::from_slice(bytes).map_err(|e| PluginManifestError::Json(e.to_string()))?;
if raw.idea_plugin_manifest_version != 1 {
return Err(PluginManifestError::Invalid(
"ideaPluginManifestVersion must be 1".to_owned(),
));
}
if raw.display_name.trim().is_empty() {
return Err(PluginManifestError::Invalid(
"displayName is required".to_owned(),
));
}
if raw.trust_level != "full" {
return Err(PluginManifestError::Invalid(
"trustLevel must be full in v1".to_owned(),
));
}
if let Some(range) = &raw.engines.idea {
if !engine_allows(range, &self.idea_version) {
return Err(PluginManifestError::IncompatibleEngine(range.clone()));
}
}
let id = PluginId::new(raw.id).map_err(|e| PluginManifestError::Invalid(e.to_string()))?;
let version = domain::PluginVersion::new(raw.version)
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?;
let main = domain::RelativePath::new(raw.main)
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?;
if !(main.as_str().ends_with(".js") || main.as_str().ends_with(".mjs")) {
return Err(PluginManifestError::Invalid(
"main must point to a .js or .mjs file".to_owned(),
));
}
let icon = raw
.icon
.map(domain::RelativePath::new)
.transpose()
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?;
let capabilities = raw
.capabilities
.into_iter()
.map(|c| match c.as_str() {
"ui" => Ok(domain::PluginCapability::Ui),
"mcp" => Ok(domain::PluginCapability::Mcp),
"tooling" => Ok(domain::PluginCapability::Tooling),
_ => Err(PluginManifestError::Invalid(format!(
"unknown capability: {c}"
))),
})
.collect::<Result<Vec<_>, _>>()?;
let contributes = validate_contributes(raw.contributes)?;
Ok(PluginManifest {
idea_plugin_manifest_version: 1,
id,
display_name: raw.display_name,
publisher: raw.publisher,
version,
description: raw.description,
engine_idea: raw.engines.idea,
main,
icon,
trust_level: PluginTrustLevel::Full,
capabilities,
activation_scope: raw.activation_scope,
contributes,
})
}
}
fn validate_contributes(raw: RawContributes) -> Result<PluginContributionSet, PluginManifestError> {
let mut seen = HashSet::new();
let mut insert = |id: &str| {
if !seen.insert(id.to_owned()) {
Err(PluginManifestError::Invalid(format!(
"duplicate contribution id: {id}"
)))
} else {
Ok(())
}
};
let menus = raw
.menus
.into_iter()
.map(|m| {
insert(&m.id)?;
if !m.top_level {
return Err(PluginManifestError::Invalid(
"menus[].topLevel must be true".to_owned(),
));
}
Ok(domain::PluginTopLevelMenuContribution {
id: m.id,
label: m.label,
top_level: true,
order: m.order,
icon: m
.icon
.map(domain::RelativePath::new)
.transpose()
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?,
})
})
.collect::<Result<Vec<_>, _>>()?;
let menu_items = raw
.menu_items
.into_iter()
.map(|m| {
insert(&m.id)?;
Ok(domain::PluginMenuItemContribution {
id: m.id,
target_menu_id: m.target_menu_id,
label: m.label,
command: domain::PluginCommandId::new(m.command)
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?,
order: m.order,
icon: m
.icon
.map(domain::RelativePath::new)
.transpose()
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?,
when: m.when,
})
})
.collect::<Result<Vec<_>, _>>()?;
let layouts = raw
.layouts
.into_iter()
.map(|l| {
insert(&l.layout_type)?;
if l.component.trim().is_empty() {
return Err(PluginManifestError::Invalid(
"layouts[].component is required".to_owned(),
));
}
Ok(domain::PluginLayoutContribution {
layout_type: domain::PluginLayoutType::new(l.layout_type)
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?,
label: l.label,
component: l.component,
order: l.order,
icon: l
.icon
.map(domain::RelativePath::new)
.transpose()
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?,
when: l.when,
})
})
.collect::<Result<Vec<_>, _>>()?;
let mcp_servers = raw
.mcp_servers
.into_iter()
.map(|s| {
insert(&s.id)?;
if s.transport != "stdio" {
return Err(PluginManifestError::Invalid(
"mcpServers[].transport must be stdio".to_owned(),
));
}
if !s.allow_absolute_command && looks_absolute(&s.command) {
return Err(PluginManifestError::Invalid(
"absolute MCP command requires allowAbsoluteCommand=true".to_owned(),
));
}
if !looks_absolute(&s.command) {
domain::RelativePath::new(s.command.clone())
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?;
}
if let Some(cwd) = &s.cwd {
if cwd != "${pluginRoot}"
&& !cwd.contains("${appDataDir}")
&& !cwd.contains("${pluginRoot}")
{
domain::RelativePath::new(cwd.clone())
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?;
}
}
Ok(domain::PluginMcpServerContribution {
id: domain::PluginMcpServerId::new(s.id)
.map_err(|e| PluginManifestError::Invalid(e.to_string()))?,
display_name: s.display_name,
command: s.command,
args: s.args,
env: s.env.into_iter().collect(),
cwd: s.cwd,
transport: s.transport,
auto_start: s.auto_start,
allow_absolute_command: s.allow_absolute_command,
})
})
.collect::<Result<Vec<_>, _>>()?;
Ok(PluginContributionSet {
menus,
menu_items,
layouts,
mcp_servers,
})
}
fn looks_absolute(path: &str) -> bool {
path.starts_with('/') || path.starts_with('\\') || path.as_bytes().get(1) == Some(&b':')
}
fn engine_allows(range: &str, current: &str) -> bool {
let cur = parse_version_tuple(current).unwrap_or((0, 0, 0));
range.split_whitespace().all(|part| {
if let Some(v) = part.strip_prefix(">=") {
parse_version_tuple(v).is_some_and(|min| cur >= min)
} else if let Some(v) = part.strip_prefix('>') {
parse_version_tuple(v).is_some_and(|min| cur > min)
} else if let Some(v) = part.strip_prefix("<=") {
parse_version_tuple(v).is_some_and(|max| cur <= max)
} else if let Some(v) = part.strip_prefix('<') {
parse_version_tuple(v).is_some_and(|max| cur < max)
} else if let Some(v) = part.strip_prefix('=') {
parse_version_tuple(v).is_some_and(|eq| cur == eq)
} else {
true
}
})
}
fn parse_version_tuple(raw: &str) -> Option<(u64, u64, u64)> {
let core = raw.split_once('-').map_or(raw, |(a, _)| a);
let mut parts = core.split('.');
Some((
parts.next()?.parse().ok()?,
parts.next()?.parse().ok()?,
parts.next()?.parse().ok()?,
))
}
#[cfg(test)]
mod tests {
use super::*;
use domain::ports::{
BackgroundCompletionStream, BackgroundTaskHandle, BackgroundTaskPortError,
BackgroundTaskRunner, BackgroundTaskSpec, EventStream, FileMetadata, IdGenerator,
PluginPackageStore, PluginRegistryStore, PluginStorageError, PluginStorageStore,
PluginStoreError, StoreError,
};
use domain::remote::RemoteRef;
use domain::{BackgroundTaskState, ProjectPath};
use std::collections::{HashMap, VecDeque};
use std::sync::Mutex;
fn validator() -> JsonPluginManifestValidator {
JsonPluginManifestValidator::new("0.3.0")
}
fn valid_manifest() -> Vec<u8> {
br#"{
"ideaPluginManifestVersion": 1,
"id": "dev.acme.gitgraph",
"displayName": "Git Graph",
"publisher": "Acme",
"version": "1.2.3",
"engines": {"idea": ">=0.1.0 <1.0.0"},
"main": "dist/index.js",
"trustLevel": "full",
"capabilities": ["ui", "mcp", "tooling"],
"contributes": {
"menus": [{"id":"dev.acme.menu","label":"Graph","topLevel":true}],
"menuItems": [{"id":"dev.acme.open","targetMenuId":"panels","label":"Open","command":"dev.acme.open"}],
"layouts": [{"type":"dev.acme.layout","label":"Graph","component":"Graph"}],
"mcpServers": [{"id":"dev.acme.mcp","displayName":"Tools","command":"servers/tool","transport":"stdio","autoStart":true}]
}
}"#.to_vec()
}
fn plugin_id() -> PluginId {
PluginId::new("dev.acme.gitgraph").unwrap()
}
fn content_hash(raw: &str) -> ContentHash {
ContentHash::new(raw).unwrap()
}
struct FakePackages {
manifests: Mutex<HashMap<String, Vec<u8>>>,
staged_manifest: Vec<u8>,
staged: Mutex<VecDeque<StagedPluginPackage>>,
removed: Mutex<Vec<String>>,
}
impl FakePackages {
fn with_manifest(bytes: Vec<u8>) -> Self {
Self::with_manifest_and_staged_count(bytes, 1)
}
fn with_manifest_and_staged_count(bytes: Vec<u8>, staged_count: usize) -> Self {
let mut manifests = HashMap::new();
manifests.insert("dev.acme.gitgraph".to_owned(), bytes);
let staged = (0..staged_count)
.map(|index| StagedPluginPackage {
root: format!("/stage/plugin-{index}"),
source: PluginInstallSource::Directory {
path_label: "/source/plugin".to_owned(),
},
content_hash: content_hash("abc123"),
})
.collect();
Self {
manifests: Mutex::new(manifests),
staged_manifest: valid_manifest(),
staged: Mutex::new(staged),
removed: Mutex::new(Vec::new()),
}
}
}
#[async_trait::async_trait]
impl PluginPackageStore for FakePackages {
async fn list_installed(&self) -> Result<Vec<domain::PluginPackageRef>, PluginStoreError> {
Ok(self
.manifests
.lock()
.unwrap()
.keys()
.map(|id| domain::PluginPackageRef {
plugin_id: Some(PluginId::new(id.clone()).unwrap()),
root: format!("/installed/{id}"),
})
.collect())
}
async fn read_manifest(
&self,
package: &domain::PluginPackageRef,
) -> Result<PluginManifestBytes, PluginStoreError> {
let key = package
.plugin_id
.as_ref()
.map_or("dev.acme.gitgraph", PluginId::as_str);
if package.plugin_id.is_none() {
return Ok(PluginManifestBytes {
bytes: self.staged_manifest.clone(),
});
}
self.manifests
.lock()
.unwrap()
.get(key)
.cloned()
.map(|bytes| PluginManifestBytes { bytes })
.ok_or(PluginStoreError::NotFound)
}
async fn install_from_archive(
&self,
_archive: &LocalPath,
) -> Result<StagedPluginPackage, PluginStoreError> {
self.staged
.lock()
.unwrap()
.pop_front()
.ok_or_else(|| PluginStoreError::Invalid("missing staged package".to_owned()))
}
async fn install_from_directory(
&self,
_dir: &LocalPath,
) -> Result<StagedPluginPackage, PluginStoreError> {
self.staged
.lock()
.unwrap()
.pop_front()
.ok_or_else(|| PluginStoreError::Invalid("missing staged package".to_owned()))
}
async fn commit_install(
&self,
staged: StagedPluginPackage,
plugin_id: &PluginId,
) -> Result<domain::PluginPackageRef, PluginStoreError> {
self.manifests
.lock()
.unwrap()
.insert(plugin_id.as_str().to_owned(), self.staged_manifest.clone());
Ok(domain::PluginPackageRef {
plugin_id: Some(plugin_id.clone()),
root: staged.root,
})
}
async fn remove_package(
&self,
plugin_id: &PluginId,
) -> Result<RemovalOutcome, PluginStoreError> {
self.removed
.lock()
.unwrap()
.push(plugin_id.as_str().to_owned());
self.manifests.lock().unwrap().remove(plugin_id.as_str());
Ok(RemovalOutcome::Removed)
}
fn bundle_url(
&self,
plugin_id: &PluginId,
entry: &domain::RelativePath,
hash: &ContentHash,
) -> Result<domain::PluginBundleUrl, PluginStoreError> {
Ok(domain::PluginBundleUrl::new(format!(
"idea-plugin://{}/current/{}/{}",
plugin_id.as_str(),
hash.as_str(),
entry.as_str()
)))
}
fn app_data_dir_label(&self) -> Option<String> {
Some("/app-data".to_owned())
}
}
#[derive(Default)]
struct FakeRegistry {
registry: Mutex<domain::PluginRegistry>,
}
#[async_trait::async_trait]
impl PluginRegistryStore for FakeRegistry {
async fn load_registry(&self) -> Result<domain::PluginRegistry, PluginRegistryError> {
Ok(self.registry.lock().unwrap().clone())
}
async fn save_registry(
&self,
registry: &domain::PluginRegistry,
) -> Result<(), PluginRegistryError> {
*self.registry.lock().unwrap() = registry.clone();
Ok(())
}
}
#[derive(Default)]
struct FakeStorage {
values: Mutex<HashMap<(String, String), serde_json::Value>>,
purged: Mutex<Vec<String>>,
}
#[async_trait::async_trait]
impl PluginStorageStore for FakeStorage {
async fn get(
&self,
plugin_id: &PluginId,
key: &str,
) -> Result<Option<serde_json::Value>, PluginStorageError> {
Ok(self
.values
.lock()
.unwrap()
.get(&(plugin_id.as_str().to_owned(), key.to_owned()))
.cloned())
}
async fn set(
&self,
plugin_id: &PluginId,
key: &str,
value: serde_json::Value,
) -> Result<(), PluginStorageError> {
self.values
.lock()
.unwrap()
.insert((plugin_id.as_str().to_owned(), key.to_owned()), value);
Ok(())
}
async fn delete(
&self,
plugin_id: &PluginId,
key: &str,
) -> Result<bool, PluginStorageError> {
Ok(self
.values
.lock()
.unwrap()
.remove(&(plugin_id.as_str().to_owned(), key.to_owned()))
.is_some())
}
async fn purge_plugin(
&self,
plugin_id: &PluginId,
) -> Result<RemovalOutcome, PluginStorageError> {
self.purged
.lock()
.unwrap()
.push(plugin_id.as_str().to_owned());
self.values
.lock()
.unwrap()
.retain(|(id, _), _| id != plugin_id.as_str());
Ok(RemovalOutcome::Removed)
}
}
#[derive(Default)]
struct FakeEvents {
events: Mutex<Vec<DomainEvent>>,
}
impl EventBus for FakeEvents {
fn publish(&self, event: DomainEvent) {
self.events.lock().unwrap().push(event);
}
fn subscribe(&self) -> EventStream {
Box::new(std::iter::empty())
}
}
#[derive(Default)]
struct FakeMcp {
reconciles: Mutex<Vec<Vec<PluginMcpServerSpec>>>,
stops: Mutex<Vec<String>>,
}
#[async_trait::async_trait]
impl PluginMcpSupervisor for FakeMcp {
async fn reconcile(
&self,
active_servers: Vec<PluginMcpServerSpec>,
) -> Result<domain::PluginMcpStatusSet, PluginMcpError> {
self.reconciles.lock().unwrap().push(active_servers.clone());
Ok(domain::PluginMcpStatusSet {
servers: active_servers
.into_iter()
.map(|spec| domain::PluginMcpStatus {
identity: spec.identity,
running: true,
error: None,
})
.collect(),
})
}
async fn stop_plugin(&self, plugin_id: &PluginId) -> Result<(), PluginMcpError> {
self.stops
.lock()
.unwrap()
.push(plugin_id.as_str().to_owned());
Ok(())
}
}
fn registry_with(state: PluginLifecycleState) -> domain::PluginRegistry {
domain::PluginRegistry {
version: 1,
plugins: vec![PluginRegistryEntry {
id: plugin_id(),
lifecycle_state: state,
source: PluginInstallSource::Directory {
path_label: "/source/plugin".to_owned(),
},
content_hash: content_hash("abc123"),
restart_required: false,
error: None,
}],
}
}
#[test]
fn validates_manifest_v1() {
let m = validator()
.validate(
&valid_manifest(),
&domain::PluginPackageRef {
plugin_id: None,
root: "x".into(),
},
)
.unwrap();
assert_eq!(m.id.as_str(), "dev.acme.gitgraph");
assert_eq!(
m.capabilities,
vec![
domain::PluginCapability::Ui,
domain::PluginCapability::Mcp,
domain::PluginCapability::Tooling,
]
);
assert_eq!(m.contributes.layouts.len(), 1);
assert_eq!(m.contributes.mcp_servers.len(), 1);
}
#[test]
fn rejects_unknown_manifest_capability() {
let mut value: serde_json::Value = serde_json::from_slice(&valid_manifest()).unwrap();
value["capabilities"] = serde_json::json!(["ui", "android"]);
let err = validator()
.validate(
&serde_json::to_vec(&value).unwrap(),
&domain::PluginPackageRef {
plugin_id: None,
root: "x".into(),
},
)
.unwrap_err();
assert_eq!(
err,
PluginManifestError::Invalid("unknown capability: android".to_owned())
);
}
#[test]
fn rejects_unsafe_main_path_and_non_full_trust() {
let mut value: serde_json::Value = serde_json::from_slice(&valid_manifest()).unwrap();
value["main"] = serde_json::json!("../dist/index.js");
assert!(validator()
.validate(
&serde_json::to_vec(&value).unwrap(),
&domain::PluginPackageRef {
plugin_id: None,
root: "x".into()
}
)
.is_err());
value["main"] = serde_json::json!("dist/index.js");
value["trustLevel"] = serde_json::json!("sandbox");
assert!(validator()
.validate(
&serde_json::to_vec(&value).unwrap(),
&domain::PluginPackageRef {
plugin_id: None,
root: "x".into()
}
)
.is_err());
}
#[test]
fn rejects_duplicate_contribution_ids() {
let mut value: serde_json::Value = serde_json::from_slice(&valid_manifest()).unwrap();
value["contributes"]["layouts"][0]["type"] = serde_json::json!("dev.acme.open");
assert!(validator()
.validate(
&serde_json::to_vec(&value).unwrap(),
&domain::PluginPackageRef {
plugin_id: None,
root: "x".into()
}
)
.is_err());
}
#[tokio::test]
async fn runtime_catalog_excludes_disabled_and_pending_uninstall_plugins() {
for state in [
PluginLifecycleState::Disabled,
PluginLifecycleState::PendingUninstall,
PluginLifecycleState::Invalid,
] {
let packages = Arc::new(FakePackages::with_manifest(valid_manifest()));
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(state)),
});
let usecase =
ListPluginRuntimeContributions::new(packages, registry, Arc::new(validator()));
let catalog = usecase.execute().await.unwrap();
assert!(catalog.plugins.is_empty(), "{state:?} must not be active");
}
}
#[tokio::test]
async fn runtime_catalog_carries_manifest_capabilities() {
let packages = Arc::new(FakePackages::with_manifest(valid_manifest()));
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Enabled)),
});
let usecase =
ListPluginRuntimeContributions::new(packages, registry, Arc::new(validator()));
let catalog = usecase.execute().await.unwrap();
assert_eq!(catalog.plugins.len(), 1);
assert_eq!(
catalog.plugins[0].capabilities,
vec![
domain::PluginCapability::Ui,
domain::PluginCapability::Mcp,
domain::PluginCapability::Tooling,
]
);
}
#[tokio::test]
async fn runtime_catalog_marks_invalid_active_plugin_and_keeps_bootstrap_alive() {
let packages = Arc::new(FakePackages::with_manifest(br#"{"broken":true}"#.to_vec()));
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Enabled)),
});
let usecase =
ListPluginRuntimeContributions::new(packages, registry.clone(), Arc::new(validator()));
let catalog = usecase.execute().await.unwrap();
assert!(
catalog.plugins.is_empty(),
"invalid active plugin must be excluded from runtime catalog"
);
let saved = registry.load_registry().await.unwrap();
let entry = saved.find(&plugin_id()).unwrap();
assert_eq!(entry.lifecycle_state, PluginLifecycleState::Invalid);
assert!(
entry
.error
.as_deref()
.unwrap_or_default()
.contains("not servable"),
"registry must carry a confined runtime error: {entry:?}"
);
}
#[tokio::test]
async fn reconcile_mcp_uses_only_enabled_auto_start_servers_with_plugin_identity() {
let packages = Arc::new(FakePackages::with_manifest(valid_manifest()));
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Enabled)),
});
let mcp = Arc::new(FakeMcp::default());
let usecase =
ReconcilePluginMcpServers::new(packages, registry, Arc::new(validator()), mcp.clone());
let statuses = usecase.execute().await.unwrap();
assert_eq!(statuses.servers.len(), 1);
assert_eq!(
statuses.servers[0].identity,
"plugin:dev.acme.gitgraph:dev.acme.mcp"
);
let reconciles = mcp.reconciles.lock().unwrap();
assert_eq!(reconciles.len(), 1);
assert_eq!(
reconciles[0][0].command,
"/installed/dev.acme.gitgraph/servers/tool"
);
assert_eq!(reconciles[0][0].cwd, "/installed/dev.acme.gitgraph");
}
#[tokio::test]
async fn reconcile_mcp_does_not_spawn_pending_uninstall_plugin_servers() {
let packages = Arc::new(FakePackages::with_manifest(valid_manifest()));
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::PendingUninstall)),
});
let mcp = Arc::new(FakeMcp::default());
let usecase =
ReconcilePluginMcpServers::new(packages, registry, Arc::new(validator()), mcp.clone());
let statuses = usecase.execute().await.unwrap();
assert!(statuses.servers.is_empty());
let reconciles = mcp.reconciles.lock().unwrap();
assert_eq!(reconciles.len(), 1);
assert!(reconciles[0].is_empty());
}
#[tokio::test]
async fn reconcile_mcp_marks_invalid_active_plugin_and_keeps_reconcile_alive() {
let packages = Arc::new(FakePackages::with_manifest(br#"{"broken":true}"#.to_vec()));
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Enabled)),
});
let mcp = Arc::new(FakeMcp::default());
let usecase = ReconcilePluginMcpServers::new(
packages,
registry.clone(),
Arc::new(validator()),
mcp.clone(),
);
let statuses = usecase.execute().await.unwrap();
assert!(statuses.servers.is_empty());
assert_eq!(mcp.reconciles.lock().unwrap().len(), 1);
assert!(mcp.reconciles.lock().unwrap()[0].is_empty());
let saved = registry.load_registry().await.unwrap();
let entry = saved.find(&plugin_id()).unwrap();
assert_eq!(entry.lifecycle_state, PluginLifecycleState::Invalid);
assert!(
entry
.error
.as_deref()
.unwrap_or_default()
.contains("MCP servers disabled"),
"registry must carry a confined MCP error: {entry:?}"
);
}
#[tokio::test]
async fn reconcile_mcp_substitutes_app_data_dir_in_plugin_server_specs() {
let mut manifest: serde_json::Value = serde_json::from_slice(&valid_manifest()).unwrap();
manifest["contributes"]["mcpServers"][0]["command"] =
serde_json::json!("${appDataDir}/plugin-tools/gitgraph");
manifest["contributes"]["mcpServers"][0]["args"] =
serde_json::json!(["--cache", "${appDataDir}/cache", "--root", "${pluginRoot}"]);
manifest["contributes"]["mcpServers"][0]["env"] = serde_json::json!({
"PLUGIN_CACHE": "${appDataDir}/cache/dev.acme.gitgraph",
"PLUGIN_ROOT": "${pluginRoot}"
});
manifest["contributes"]["mcpServers"][0]["cwd"] =
serde_json::json!("${appDataDir}/work/dev.acme.gitgraph");
let packages = Arc::new(FakePackages::with_manifest(
serde_json::to_vec(&manifest).unwrap(),
));
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Enabled)),
});
let mcp = Arc::new(FakeMcp::default());
let usecase =
ReconcilePluginMcpServers::new(packages, registry, Arc::new(validator()), mcp.clone());
usecase.execute().await.unwrap();
let reconciles = mcp.reconciles.lock().unwrap();
let spec = &reconciles[0][0];
assert_eq!(spec.command, "/app-data/plugin-tools/gitgraph");
assert_eq!(
spec.args,
vec![
"--cache".to_owned(),
"/app-data/cache".to_owned(),
"--root".to_owned(),
"/installed/dev.acme.gitgraph".to_owned()
]
);
assert!(spec.env.contains(&(
"PLUGIN_CACHE".to_owned(),
"/app-data/cache/dev.acme.gitgraph".to_owned()
)));
assert!(spec.env.contains(&(
"PLUGIN_ROOT".to_owned(),
"/installed/dev.acme.gitgraph".to_owned()
)));
assert_eq!(spec.cwd, "/app-data/work/dev.acme.gitgraph");
}
#[tokio::test]
async fn disable_stops_plugin_and_removes_it_from_runtime_catalog() {
let packages = Arc::new(FakePackages::with_manifest(valid_manifest()));
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Enabled)),
});
let events = Arc::new(FakeEvents::default());
let mcp = Arc::new(FakeMcp::default());
let disable = SetPluginEnabled::new(
packages.clone(),
registry.clone(),
Arc::new(validator()),
events.clone(),
mcp.clone(),
);
let admin = disable
.execute(SetPluginEnabledInput {
plugin_id: "dev.acme.gitgraph".to_owned(),
enabled: false,
})
.await
.unwrap();
assert!(!admin.enabled);
assert_eq!(admin.lifecycle_state, PluginLifecycleState::Disabled);
assert_eq!(&*mcp.stops.lock().unwrap(), &["dev.acme.gitgraph"]);
assert!(events.events.lock().unwrap().iter().any(|event| matches!(
event,
DomainEvent::PluginDisabled {
restart_required: true,
..
}
)));
let runtime =
ListPluginRuntimeContributions::new(packages, registry, Arc::new(validator()))
.execute()
.await
.unwrap();
assert!(runtime.plugins.is_empty());
}
#[tokio::test]
async fn uninstall_removes_registry_package_and_stops_mcp() {
let packages = Arc::new(FakePackages::with_manifest(valid_manifest()));
let storage = Arc::new(FakeStorage::default());
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Enabled)),
});
let events = Arc::new(FakeEvents::default());
let mcp = Arc::new(FakeMcp::default());
let uninstall = UninstallPlugin::new(
packages.clone(),
storage.clone(),
registry.clone(),
events.clone(),
mcp.clone(),
);
let result = uninstall
.execute(UninstallPluginInput {
plugin_id: "dev.acme.gitgraph".to_owned(),
})
.await
.unwrap();
assert_eq!(result.removal_outcome, RemovalOutcome::Removed);
assert!(result.restart_required);
assert!(registry.load_registry().await.unwrap().plugins.is_empty());
assert_eq!(&*packages.removed.lock().unwrap(), &["dev.acme.gitgraph"]);
assert_eq!(&*storage.purged.lock().unwrap(), &["dev.acme.gitgraph"]);
assert_eq!(&*mcp.stops.lock().unwrap(), &["dev.acme.gitgraph"]);
assert!(events.events.lock().unwrap().iter().any(|event| matches!(
event,
DomainEvent::PluginUninstalled {
restart_required: true,
..
}
)));
}
#[tokio::test]
async fn plugin_storage_round_trips_json_for_runtime_active_plugin() {
let storage = Arc::new(FakeStorage::default());
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Enabled)),
});
let access = PluginStorageAccess::new(storage, registry);
access
.set(PluginStorageSetInput {
plugin_id: "dev.acme.gitgraph".to_owned(),
key: "helloPlugin.launches".to_owned(),
value: serde_json::json!({"count": 2}),
})
.await
.unwrap();
let value = access
.get(PluginStorageGetInput {
plugin_id: "dev.acme.gitgraph".to_owned(),
key: "helloPlugin.launches".to_owned(),
})
.await
.unwrap();
assert_eq!(value, Some(serde_json::json!({"count": 2})));
assert!(access
.delete(PluginStorageGetInput {
plugin_id: "dev.acme.gitgraph".to_owned(),
key: "helloPlugin.launches".to_owned(),
})
.await
.unwrap());
assert_eq!(
access
.get(PluginStorageGetInput {
plugin_id: "dev.acme.gitgraph".to_owned(),
key: "helloPlugin.launches".to_owned(),
})
.await
.unwrap(),
None
);
}
#[tokio::test]
async fn plugin_storage_rejects_inactive_plugin_and_invalid_key() {
let storage = Arc::new(FakeStorage::default());
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Disabled)),
});
let access = PluginStorageAccess::new(storage, registry);
let inactive = access
.set(PluginStorageSetInput {
plugin_id: "dev.acme.gitgraph".to_owned(),
key: "helloPlugin.launches".to_owned(),
value: serde_json::json!(1),
})
.await
.unwrap_err();
assert_eq!(
inactive,
AppError::Invalid("plugin is not runtime-active".to_owned())
);
let storage = Arc::new(FakeStorage::default());
let registry = Arc::new(FakeRegistry {
registry: Mutex::new(registry_with(PluginLifecycleState::Enabled)),
});
let access = PluginStorageAccess::new(storage, registry);
let invalid = access
.get(PluginStorageGetInput {
plugin_id: "dev.acme.gitgraph".to_owned(),
key: " ".to_owned(),
})
.await
.unwrap_err();
assert_eq!(
invalid,
AppError::Invalid("plugin storage key must not be empty".to_owned())
);
}
#[tokio::test]
async fn uninstall_then_reinstall_leaves_runtime_catalog_active_without_residue() {
let packages = Arc::new(FakePackages::with_manifest_and_staged_count(
valid_manifest(),
2,
));
let registry = Arc::new(FakeRegistry::default());
let events = Arc::new(FakeEvents::default());
let mcp = Arc::new(FakeMcp::default());
let storage = Arc::new(FakeStorage::default());
let install = InstallPluginFromDirectory::new(
packages.clone(),
registry.clone(),
Arc::new(validator()),
events.clone(),
mcp.clone(),
);
let uninstall = UninstallPlugin::new(
packages.clone(),
storage.clone(),
registry.clone(),
events,
mcp.clone(),
);
install.execute("/source/plugin".to_owned()).await.unwrap();
uninstall
.execute(UninstallPluginInput {
plugin_id: "dev.acme.gitgraph".to_owned(),
})
.await
.unwrap();
assert!(registry.load_registry().await.unwrap().plugins.is_empty());
let result = install.execute("/source/plugin".to_owned()).await.unwrap();
let runtime = ListPluginRuntimeContributions::new(
packages.clone(),
registry.clone(),
Arc::new(validator()),
)
.execute()
.await
.unwrap();
assert_eq!(result.plugin.lifecycle_state, PluginLifecycleState::Enabled);
assert_eq!(runtime.plugins.len(), 1);
assert_eq!(runtime.plugins[0].id, "dev.acme.gitgraph");
let saved = registry.load_registry().await.unwrap();
let entry = saved.find(&plugin_id()).unwrap();
assert_eq!(entry.lifecycle_state, PluginLifecycleState::Enabled);
assert!(entry.error.is_none());
assert_eq!(&*packages.removed.lock().unwrap(), &["dev.acme.gitgraph"]);
assert_eq!(&*mcp.stops.lock().unwrap(), &["dev.acme.gitgraph"]);
}
#[derive(Default)]
struct FakeProjectStore {
projects: Mutex<HashMap<ProjectId, Project>>,
}
#[async_trait::async_trait]
impl ProjectStore for FakeProjectStore {
async fn list_projects(&self) -> Result<Vec<Project>, StoreError> {
Ok(self.projects.lock().unwrap().values().cloned().collect())
}
async fn load_project(&self, id: ProjectId) -> Result<Project, StoreError> {
self.projects
.lock()
.unwrap()
.get(&id)
.cloned()
.ok_or(StoreError::NotFound)
}
async fn save_project(&self, project: &Project) -> Result<(), StoreError> {
self.projects
.lock()
.unwrap()
.insert(project.id, project.clone());
Ok(())
}
async fn save_workspace(&self, _workspace: &domain::Workspace) -> Result<(), StoreError> {
Ok(())
}
async fn load_workspace(&self) -> Result<domain::Workspace, StoreError> {
Ok(domain::Workspace::default())
}
}
#[derive(Default)]
struct FakeWorkspaceFs {
files: Mutex<HashMap<String, Vec<u8>>>,
}
impl FakeWorkspaceFs {
fn seed_file(&self, path: &str, bytes: impl Into<Vec<u8>>) {
self.files
.lock()
.unwrap()
.insert(path.to_owned(), bytes.into());
}
}
#[derive(Default)]
struct FakeBackgroundTaskStore {
tasks: Mutex<HashMap<TaskId, BackgroundTask>>,
}
impl FakeBackgroundTaskStore {
fn task(&self, task_id: TaskId) -> Option<BackgroundTask> {
self.tasks.lock().unwrap().get(&task_id).cloned()
}
}
#[async_trait::async_trait]
impl BackgroundTaskStore for FakeBackgroundTaskStore {
async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
let mut tasks = self.tasks.lock().unwrap();
if tasks.contains_key(&task.id) {
return Err(BackgroundTaskPortError::AlreadyExists);
}
tasks.insert(task.id, task.clone());
Ok(())
}
async fn get(&self, id: TaskId) -> Result<Option<BackgroundTask>, BackgroundTaskPortError> {
Ok(self.task(id))
}
async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> {
self.tasks.lock().unwrap().insert(task.id, task.clone());
Ok(())
}
async fn list_open_for_agent(
&self,
agent_id: AgentId,
) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
Ok(self
.tasks
.lock()
.unwrap()
.values()
.filter(|task| task.owner_agent_id == agent_id && !task.is_terminal())
.cloned()
.collect())
}
async fn list_undelivered_completions(
&self,
) -> Result<Vec<BackgroundTask>, BackgroundTaskPortError> {
Ok(self
.tasks
.lock()
.unwrap()
.values()
.filter(|task| task.has_pending_completion_delivery())
.cloned()
.collect())
}
async fn mark_completion_delivered(
&self,
task_id: TaskId,
) -> Result<(), BackgroundTaskPortError> {
let task = self
.task(task_id)
.ok_or(BackgroundTaskPortError::NotFound)?
.mark_completion_delivered()
.map_err(|err| BackgroundTaskPortError::Invalid(err.to_string()))?;
self.save(&task).await
}
}
#[derive(Default)]
struct FakeBackgroundTaskRunner {
specs: Mutex<Vec<BackgroundTaskSpec>>,
}
impl FakeBackgroundTaskRunner {
fn specs(&self) -> Vec<BackgroundTaskSpec> {
self.specs.lock().unwrap().clone()
}
}
#[async_trait::async_trait]
impl BackgroundTaskRunner for FakeBackgroundTaskRunner {
async fn spawn(
&self,
spec: BackgroundTaskSpec,
) -> Result<BackgroundTaskHandle, BackgroundTaskPortError> {
let task_id = spec.task_id;
self.specs.lock().unwrap().push(spec);
Ok(BackgroundTaskHandle { task_id })
}
async fn cancel(&self, _task_id: TaskId) -> Result<(), BackgroundTaskPortError> {
Ok(())
}
fn subscribe_completions(&self) -> BackgroundCompletionStream {
Box::new(std::iter::empty())
}
}
struct FixedClock(i64);
impl domain::ports::Clock for FixedClock {
fn now_millis(&self) -> i64 {
self.0
}
}
struct FixedIds(Uuid);
impl IdGenerator for FixedIds {
fn new_uuid(&self) -> Uuid {
self.0
}
}
#[derive(Default)]
struct FakeProcessSpawner {
outputs: Mutex<HashMap<String, Result<Output, ProcessError>>>,
specs: Mutex<Vec<SpawnSpec>>,
}
impl FakeProcessSpawner {
fn seed(&self, command: &str, output: Result<Output, ProcessError>) {
self.outputs
.lock()
.unwrap()
.insert(command.to_owned(), output);
}
fn specs(&self) -> Vec<SpawnSpec> {
self.specs.lock().unwrap().clone()
}
}
#[async_trait::async_trait]
impl ProcessSpawner for FakeProcessSpawner {
async fn run(&self, spec: SpawnSpec) -> Result<Output, ProcessError> {
self.specs.lock().unwrap().push(spec.clone());
self.outputs
.lock()
.unwrap()
.get(&spec.command)
.cloned()
.unwrap_or_else(|| Err(ProcessError::Spawn(format!("{} not found", spec.command))))
}
}
#[derive(Default)]
struct FakeEnvironmentReader {
values: Mutex<HashMap<String, String>>,
}
impl FakeEnvironmentReader {
fn set(&self, name: &str, value: &str) {
self.values
.lock()
.unwrap()
.insert(name.to_owned(), value.to_owned());
}
}
impl EnvironmentReader for FakeEnvironmentReader {
fn get(&self, name: &str) -> Option<String> {
self.values.lock().unwrap().get(name).cloned()
}
}
#[async_trait::async_trait]
impl FileSystem for FakeWorkspaceFs {
async fn read(&self, path: &RemotePath) -> Result<Vec<u8>, domain::ports::FsError> {
self.files
.lock()
.unwrap()
.get(path.as_str())
.cloned()
.ok_or_else(|| domain::ports::FsError::NotFound(path.as_str().to_owned()))
}
async fn write(
&self,
path: &RemotePath,
data: &[u8],
) -> Result<(), domain::ports::FsError> {
self.files
.lock()
.unwrap()
.insert(path.as_str().to_owned(), data.to_vec());
Ok(())
}
async fn exists(&self, path: &RemotePath) -> Result<bool, domain::ports::FsError> {
let files = self.files.lock().unwrap();
Ok(files.contains_key(path.as_str())
|| files
.keys()
.any(|p| p.starts_with(&format!("{}/", path.as_str()))))
}
async fn metadata(
&self,
path: &RemotePath,
) -> Result<FileMetadata, domain::ports::FsError> {
let files = self.files.lock().unwrap();
if let Some(bytes) = files.get(path.as_str()) {
return Ok(FileMetadata {
is_file: true,
is_dir: false,
len: Some(bytes.len() as u64),
});
}
if files
.keys()
.any(|p| p.starts_with(&format!("{}/", path.as_str().trim_end_matches('/'))))
{
return Ok(FileMetadata {
is_file: false,
is_dir: true,
len: None,
});
}
Err(domain::ports::FsError::NotFound(path.as_str().to_owned()))
}
async fn create_dir_all(&self, _path: &RemotePath) -> Result<(), domain::ports::FsError> {
Ok(())
}
async fn list(&self, path: &RemotePath) -> Result<Vec<DirEntry>, domain::ports::FsError> {
let prefix = path.as_str().trim_end_matches('/');
let prefix = if prefix.is_empty() {
String::new()
} else {
format!("{prefix}/")
};
let mut seen = HashMap::<String, bool>::new();
for file in self.files.lock().unwrap().keys() {
let Some(rest) = file.strip_prefix(&prefix) else {
continue;
};
if rest.is_empty() {
continue;
}
let (name, is_dir) = match rest.split_once('/') {
Some((name, _)) => (name.to_owned(), true),
None => (rest.to_owned(), false),
};
seen.entry(name)
.and_modify(|existing| *existing |= is_dir)
.or_insert(is_dir);
}
if seen.is_empty() && !self.exists(path).await? {
return Err(domain::ports::FsError::NotFound(path.as_str().to_owned()));
}
Ok(seen
.into_iter()
.map(|(name, is_dir)| DirEntry { name, is_dir })
.collect())
}
async fn symlink(
&self,
_src: &RemotePath,
_dst: &RemotePath,
) -> Result<(), domain::ports::FsError> {
Ok(())
}
}
fn workspace_fixture() -> (ProjectId, Arc<FakeProjectStore>, Arc<FakeWorkspaceFs>) {
let project_id = ProjectId::new_random();
let project = Project::new(
project_id,
"Example",
ProjectPath::new("/workspace/example").unwrap(),
RemoteRef::Local,
0,
)
.unwrap();
let store = Arc::new(FakeProjectStore::default());
store.projects.lock().unwrap().insert(project_id, project);
(project_id, store, Arc::new(FakeWorkspaceFs::default()))
}
#[tokio::test]
async fn plugin_workspace_rejects_paths_outside_project_root() {
let (project_id, projects, fs) = workspace_fixture();
let access = PluginWorkspaceAccess::new(projects, fs);
let err = access
.read_text(PluginWorkspacePathInput {
project_id: project_id.to_string(),
path: "../secret.txt".to_owned(),
})
.await
.unwrap_err();
assert_eq!(err.code(), "INVALID");
}
#[tokio::test]
async fn plugin_workspace_reads_writes_lists_and_stats_files() {
let (project_id, projects, fs) = workspace_fixture();
let access = PluginWorkspaceAccess::new(projects, fs.clone());
access
.write_text(PluginWorkspaceWriteTextInput {
project_id: project_id.to_string(),
path: "src/main.rs".to_owned(),
content: "fn main() {}\n".to_owned(),
})
.await
.unwrap();
let read = access
.read_text(PluginWorkspacePathInput {
project_id: project_id.to_string(),
path: "src/main.rs".to_owned(),
})
.await
.unwrap();
assert_eq!(read.content, "fn main() {}\n");
let listing = access
.list_dir(PluginWorkspacePathInput {
project_id: project_id.to_string(),
path: "src".to_owned(),
})
.await
.unwrap();
assert_eq!(listing.entries[0].path, "src/main.rs");
let stat = access
.stat(PluginWorkspacePathInput {
project_id: project_id.to_string(),
path: "src/main.rs".to_owned(),
})
.await
.unwrap();
assert!(stat.exists);
assert!(stat.is_file);
assert_eq!(stat.len, Some(13));
}
#[tokio::test]
async fn plugin_event_subscriptions_receive_workspace_changes_without_raw_event_leak() {
let (project_id, projects, fs) = workspace_fixture();
let events = Arc::new(FakeEvents::default());
let access = PluginWorkspaceAccess::new(projects.clone(), fs).with_events(events.clone());
let subscriptions = PluginEventSubscriptions::new(
projects,
Arc::new(FixedIds(Uuid::from_u128(127))) as Arc<dyn IdGenerator>,
Arc::new(FixedClock(1_700_000_000_000)) as Arc<dyn Clock>,
);
let subscription = subscriptions
.subscribe(PluginEventSubscribeInput {
project_id: project_id.to_string(),
event_types: vec!["workspaceFileChanged".to_owned()],
capacity: Some(10),
})
.await
.unwrap();
access
.write_text(PluginWorkspaceWriteTextInput {
project_id: project_id.to_string(),
path: "generated.txt".to_owned(),
content: "hello\n".to_owned(),
})
.await
.unwrap();
for event in events.events.lock().unwrap().iter() {
subscriptions.record_domain_event(event);
}
subscriptions.record_domain_event(&DomainEvent::PluginInstalled {
plugin_id: plugin_id(),
version: domain::PluginVersion::new("1.0.0").unwrap(),
});
let batch = subscriptions
.poll(PluginEventPollInput {
subscription_id: subscription.subscription_id.clone(),
max_events: Some(10),
})
.unwrap();
assert_eq!(batch.dropped, 0);
assert_eq!(batch.events.len(), 1);
match &batch.events[0] {
PluginPublicEvent::WorkspaceFileChanged {
project_id: observed,
path,
operation,
occurred_at_ms,
..
} => {
assert_eq!(observed, &project_id.to_string());
assert_eq!(path, "generated.txt");
assert_eq!(operation, "changed");
assert_eq!(*occurred_at_ms, 1_700_000_000_000);
}
other => panic!("unexpected public event: {other:?}"),
}
subscriptions.unsubscribe(PluginEventUnsubscribeInput {
subscription_id: subscription.subscription_id.clone(),
});
let err = subscriptions
.poll(PluginEventPollInput {
subscription_id: subscription.subscription_id,
max_events: None,
})
.unwrap_err();
assert_eq!(err.code(), "NOT_FOUND");
}
#[tokio::test]
async fn plugin_event_subscriptions_project_task_events_with_bounded_retention() {
let (project_id, projects, _fs) = workspace_fixture();
let subscriptions = PluginEventSubscriptions::new(
projects,
Arc::new(FixedIds(Uuid::from_u128(128))) as Arc<dyn IdGenerator>,
Arc::new(FixedClock(1_700_000_000_100)) as Arc<dyn Clock>,
);
let subscription = subscriptions
.subscribe(PluginEventSubscribeInput {
project_id: project_id.to_string(),
event_types: vec!["backgroundTaskChanged".to_owned()],
capacity: Some(1),
})
.await
.unwrap();
let owner_agent_id = AgentId::from_uuid(Uuid::from_u128(88));
let task_id = TaskId::from_uuid(Uuid::from_u128(99));
subscriptions.record_domain_event(&DomainEvent::BackgroundTaskStarted {
project_id,
task_id,
owner_agent_id,
});
subscriptions.record_domain_event(&DomainEvent::BackgroundTaskCompleted {
project_id,
task_id,
owner_agent_id,
rendezvous: None,
});
let batch = subscriptions
.poll(PluginEventPollInput {
subscription_id: subscription.subscription_id,
max_events: None,
})
.unwrap();
assert_eq!(batch.dropped, 1);
assert_eq!(batch.events.len(), 1);
match &batch.events[0] {
PluginPublicEvent::BackgroundTaskChanged {
project_id: observed,
task_id: observed_task,
owner_agent_id: observed_owner,
state,
..
} => {
assert_eq!(observed, &project_id.to_string());
assert_eq!(observed_task, &task_id.to_string());
assert_eq!(observed_owner, &owner_agent_id.to_string());
assert_eq!(state, "completed");
}
other => panic!("unexpected public event: {other:?}"),
}
}
#[tokio::test]
async fn plugin_config_documents_read_and_merge_patch_json_under_project_root() {
let (project_id, projects, fs) = workspace_fixture();
fs.seed_file(
"/workspace/example/config/settings.json",
br#"{"name":"demo","enabled":false,"removeMe":true,"nested":{"keep":1}}"#.to_vec(),
);
let events = Arc::new(FakeEvents::default());
let documents =
PluginConfigDocuments::new(projects, fs.clone()).with_events(events.clone());
let read = documents
.read(PluginConfigDocumentReadInput {
project_id: project_id.to_string(),
path: "config/settings.json".to_owned(),
format: None,
})
.await
.unwrap();
assert_eq!(read.format, "json");
assert_eq!(read.value["name"], "demo");
let result = documents
.update(PluginConfigDocumentUpdateInput {
project_id: project_id.to_string(),
path: "config/settings.json".to_owned(),
format: None,
mode: Some("mergePatch".to_owned()),
value: serde_json::json!({
"enabled": true,
"removeMe": null,
"nested": {"added": 2}
}),
})
.await
.unwrap();
assert_eq!(result.mode, "mergePatch");
assert!(result.bytes_written > 0);
let updated: serde_json::Value = serde_json::from_slice(
&fs.read(&RemotePath::new("/workspace/example/config/settings.json"))
.await
.unwrap(),
)
.unwrap();
assert_eq!(updated["enabled"], true);
assert!(updated.get("removeMe").is_none());
assert_eq!(updated["nested"]["keep"], 1);
assert_eq!(updated["nested"]["added"], 2);
assert!(events.events.lock().unwrap().iter().any(|event| matches!(
event,
DomainEvent::PluginWorkspaceFileChanged { path, .. }
if path == "config/settings.json"
)));
}
#[tokio::test]
async fn plugin_config_documents_replace_json_and_reject_unsupported_formats() {
let (project_id, projects, fs) = workspace_fixture();
fs.seed_file(
"/workspace/example/config.json",
br#"{"old":true}"#.to_vec(),
);
let documents = PluginConfigDocuments::new(projects, fs.clone());
documents
.update(PluginConfigDocumentUpdateInput {
project_id: project_id.to_string(),
path: "config.json".to_owned(),
format: Some("json".to_owned()),
mode: Some("replace".to_owned()),
value: serde_json::json!({"new": true}),
})
.await
.unwrap();
let updated = documents
.read(PluginConfigDocumentReadInput {
project_id: project_id.to_string(),
path: "config.json".to_owned(),
format: Some("json".to_owned()),
})
.await
.unwrap();
assert_eq!(updated.value, serde_json::json!({"new": true}));
let err = documents
.read(PluginConfigDocumentReadInput {
project_id: project_id.to_string(),
path: "Cargo.toml".to_owned(),
format: Some("toml".to_owned()),
})
.await
.unwrap_err();
assert_eq!(err.code(), "INVALID");
assert!(err.to_string().contains("supported formats: json"));
}
#[tokio::test]
async fn query_project_structure_detects_generic_markers_and_modules() {
let (project_id, projects, fs) = workspace_fixture();
fs.seed_file("/workspace/example/Cargo.toml", b"[package]\n".to_vec());
fs.seed_file(
"/workspace/example/crates/app/Cargo.toml",
b"[package]\n".to_vec(),
);
fs.seed_file("/workspace/example/crates/app/src/lib.rs", b"".to_vec());
fs.seed_file(
"/workspace/example/node_modules/skip/package.json",
b"{}".to_vec(),
);
let query = QueryProjectStructure::new(projects, fs);
let result = query
.execute(QueryProjectStructureInput {
project_id: project_id.to_string(),
path: None,
max_depth: Some(4),
max_entries: Some(100),
})
.await
.unwrap();
assert!(result
.conventions
.iter()
.any(|c| c.id == "rust-cargo" && c.marker_path == "Cargo.toml"));
assert!(result
.modules
.iter()
.any(|m| m.path == "crates/app" && m.convention_id == "rust-cargo"));
assert!(!result
.entries
.iter()
.any(|entry| entry.path == "node_modules/skip/package.json"));
}
#[tokio::test]
async fn plugin_toolchain_diagnostics_detects_tool_env_and_file_prerequisites() {
let (project_id, projects, fs) = workspace_fixture();
fs.seed_file("/workspace/example/Cargo.toml", b"[package]\n".to_vec());
let processes = Arc::new(FakeProcessSpawner::default());
processes.seed(
"cargo",
Ok(Output {
status: domain::ports::ExitStatus { code: Some(0) },
stdout: b"cargo 1.80.0\n".to_vec(),
stderr: Vec::new(),
}),
);
let env = Arc::new(FakeEnvironmentReader::default());
env.set("RUSTUP_HOME", "/rustup");
let diagnostics = PluginToolchainDiagnostics::new(projects, fs, processes.clone(), env);
let result = diagnostics
.diagnose(PluginToolchainDiagnosticInput {
project_id: project_id.to_string(),
cwd: Some(".".to_owned()),
tools: vec![PluginToolRequirement {
id: "rust".to_owned(),
executable: "cargo".to_owned(),
version_args: vec!["--version".to_owned()],
required: true,
env: vec![("CARGO_TERM_COLOR".to_owned(), "never".to_owned())],
}],
env: vec![PluginEnvRequirement {
name: "RUSTUP_HOME".to_owned(),
required: true,
equals: None,
}],
files: vec![PluginFileRequirement {
path: "Cargo.toml".to_owned(),
required: true,
kind: Some("file".to_owned()),
}],
})
.await
.unwrap();
assert!(result.ok);
assert_eq!(result.cwd, "");
assert_eq!(result.tools[0].version.as_deref(), Some("cargo 1.80.0"));
assert_eq!(result.env[0].value.as_deref(), Some("/rustup"));
assert_eq!(result.files[0].kind, "file");
let spec = processes.specs().pop().expect("process probe captured");
assert_eq!(spec.command, "cargo");
assert_eq!(spec.cwd.as_str(), "/workspace/example");
assert_eq!(
spec.env,
vec![("CARGO_TERM_COLOR".to_owned(), "never".to_owned())]
);
}
#[tokio::test]
async fn plugin_toolchain_diagnostics_reports_missing_required_tool_without_failing_usecase() {
let (project_id, projects, fs) = workspace_fixture();
let processes = Arc::new(FakeProcessSpawner::default());
processes.seed(
"missing-tool",
Err(ProcessError::Spawn("missing-tool: not found".to_owned())),
);
let diagnostics = PluginToolchainDiagnostics::new(
projects,
fs,
processes,
Arc::new(FakeEnvironmentReader::default()),
);
let result = diagnostics
.diagnose(PluginToolchainDiagnosticInput {
project_id: project_id.to_string(),
cwd: None,
tools: vec![PluginToolRequirement {
id: "required-cli".to_owned(),
executable: "missing-tool".to_owned(),
version_args: Vec::new(),
required: true,
env: Vec::new(),
}],
env: Vec::new(),
files: Vec::new(),
})
.await
.unwrap();
assert!(!result.ok);
assert_eq!(result.tools[0].status, "missing");
assert!(!result.tools[0].present);
assert!(result
.messages
.iter()
.any(|message| message.level == "error"
&& message.message.contains("missing-tool: not found")));
}
fn plugin_tasks_fixture() -> (
ProjectId,
Arc<FakeProjectStore>,
Arc<FakeBackgroundTaskStore>,
Arc<FakeBackgroundTaskRunner>,
PluginCommandTasks,
) {
let (project_id, projects, _fs) = workspace_fixture();
let tasks = Arc::new(FakeBackgroundTaskStore::default());
let runner = Arc::new(FakeBackgroundTaskRunner::default());
let spawn = Arc::new(SpawnBackgroundCommand::new(
Arc::clone(&tasks) as Arc<dyn BackgroundTaskStore>,
Arc::clone(&runner) as Arc<dyn BackgroundTaskRunner>,
Arc::new(FixedClock(1_700_000_000_000)) as Arc<dyn domain::ports::Clock>,
Arc::new(FixedIds(Uuid::from_u128(125))) as Arc<dyn IdGenerator>,
));
let facade = PluginCommandTasks::new(
Arc::clone(&projects) as Arc<dyn ProjectStore>,
Arc::clone(&tasks) as Arc<dyn BackgroundTaskStore>,
spawn,
);
(project_id, projects, tasks, runner, facade)
}
#[tokio::test]
async fn plugin_command_tasks_launches_tracked_command_under_project_root() {
let (project_id, _projects, tasks, runner, facade) = plugin_tasks_fixture();
let owner = AgentId::from_uuid(Uuid::from_u128(77));
let task = facade
.run_command(PluginRunCommandInput {
project_id: project_id.to_string(),
owner_agent_id: owner.to_string(),
label: "cargo test".to_owned(),
command: "cargo".to_owned(),
args: vec!["test".to_owned()],
cwd: Some("crates/application".to_owned()),
env: vec![("RUST_LOG".to_owned(), "debug".to_owned())],
record_only: true,
deadline_ms: Some(1_800_000_000_000),
})
.await
.unwrap();
assert_eq!(task.id, TaskId::from_uuid(Uuid::from_u128(125)));
assert_eq!(task.project_id, project_id);
assert_eq!(task.owner_agent_id, owner);
assert_eq!(task.state, BackgroundTaskState::Running);
assert_eq!(task.wake_policy, BackgroundTaskWakePolicy::RecordOnly);
let persisted = facade
.get_status(PluginTaskStatusInput {
task_id: task.id.to_string(),
})
.await
.unwrap()
.expect("task persisted");
assert_eq!(persisted.state, BackgroundTaskState::Running);
assert_eq!(
tasks.task(task.id).unwrap().state,
BackgroundTaskState::Running
);
let spec = runner.specs().pop().expect("runner invoked");
assert_eq!(spec.task_id, task.id);
assert_eq!(spec.project_id, project_id);
assert_eq!(spec.owner_agent_id, owner);
assert_eq!(spec.wake_policy, BackgroundTaskWakePolicy::RecordOnly);
assert_eq!(spec.deadline_ms, Some(1_800_000_000_000));
let command = spec.command.expect("command spec");
assert_eq!(command.command, "cargo");
assert_eq!(command.args, vec!["test"]);
assert_eq!(
command.cwd.as_str(),
"/workspace/example/crates/application"
);
assert_eq!(
command.env,
vec![("RUST_LOG".to_owned(), "debug".to_owned())]
);
}
#[tokio::test]
async fn plugin_command_tasks_rejects_cwd_outside_project_root() {
let (project_id, _projects, tasks, runner, facade) = plugin_tasks_fixture();
let owner = AgentId::from_uuid(Uuid::from_u128(77));
let err = facade
.run_command(PluginRunCommandInput {
project_id: project_id.to_string(),
owner_agent_id: owner.to_string(),
label: "escape".to_owned(),
command: "sh".to_owned(),
args: vec!["-c".to_owned(), "pwd".to_owned()],
cwd: Some("../outside".to_owned()),
env: Vec::new(),
record_only: false,
deadline_ms: None,
})
.await
.unwrap_err();
assert_eq!(err.code(), "INVALID");
assert!(tasks.tasks.lock().unwrap().is_empty());
assert!(runner.specs().is_empty());
}
}