use std::{
path::{Path, PathBuf},
sync::Arc,
};
use tokio::sync::{Mutex, Semaphore};
use crate::{
domain::{
errors::{AgentError, AgentResult, ErrorCode},
protocol::{AgentMessage, SkillScope},
},
infrastructure::{
pi_skill_runtime::PiSkillRuntime,
skill_catalog::SkillCatalogRegistry,
skill_store::{SkillStore, WorkspacePiCommands},
},
};
const MAX_PROVIDER_REQUESTS: usize = 4;
#[derive(Clone)]
pub struct SkillService {
catalogs: Arc<SkillCatalogRegistry>,
store: Arc<SkillStore>,
runtime: PiSkillRuntime,
runtime_cwd: PathBuf,
provider_requests: Arc<Semaphore>,
installation_lock: Arc<Mutex<()>>,
}
#[derive(Debug)]
pub enum SkillRequest {
Search {
request_id: String,
provider: Option<String>,
query: String,
limit: u16,
offset: u32,
},
ListInstalled {
request_id: String,
workspace: Option<String>,
include_all: bool,
workspaces: Vec<String>,
},
Install {
request_id: String,
provider: String,
skill_id: String,
scope: SkillScope,
workspace: Option<String>,
confirm_update: bool,
},
Uninstall {
request_id: String,
installation_id: String,
},
}
impl SkillRequest {
pub(crate) fn request_id(&self) -> &str {
match self {
Self::Search { request_id, .. }
| Self::ListInstalled { request_id, .. }
| Self::Install { request_id, .. }
| Self::Uninstall { request_id, .. } => request_id,
}
}
}
impl SkillService {
pub fn new(
catalogs: Arc<SkillCatalogRegistry>,
store: Arc<SkillStore>,
runtime: PiSkillRuntime,
runtime_cwd: PathBuf,
) -> Self {
Self {
catalogs,
store,
runtime,
runtime_cwd,
provider_requests: Arc::new(Semaphore::new(MAX_PROVIDER_REQUESTS)),
installation_lock: Arc::new(Mutex::new(())),
}
}
pub async fn execute(&self, request: SkillRequest) -> AgentResult<AgentMessage> {
match request {
SkillRequest::Search {
request_id,
provider,
query,
limit,
offset,
} => {
let catalog = self.catalogs.get(provider.as_deref())?;
let provider = catalog.id().to_string();
let _permit = self.provider_permit().await?;
let result = catalog.search(&query, limit, offset).await?;
Ok(AgentMessage::SkillsSearchResults {
request_id,
provider,
query,
items: result.items,
next_offset: result.next_offset,
})
}
SkillRequest::ListInstalled {
request_id,
workspace,
include_all,
workspaces,
} => {
let additional_workspaces = if include_all { workspaces } else { Vec::new() };
let requested_workspace = workspace
.as_deref()
.map(PathBuf::from)
.unwrap_or_else(|| self.runtime_cwd.clone());
let runtime_workspaces = self.store.validated_runtime_workspaces(
Some(&requested_workspace),
&additional_workspaces,
)?;
let mut runtime_commands = Vec::with_capacity(runtime_workspaces.len());
for workspace in runtime_workspaces {
let commands = self.runtime.get_commands(&workspace).await?;
runtime_commands.push(WorkspacePiCommands {
workspace,
commands,
});
}
let store = self.store.clone();
let items = tokio::task::spawn_blocking(move || {
store.list_with_runtime_commands(
Some(&requested_workspace),
&additional_workspaces,
runtime_commands,
)
})
.await
.map_err(|_| filesystem_task_failed())??;
Ok(AgentMessage::SkillsInstalledListed { request_id, items })
}
SkillRequest::Install {
request_id,
provider,
skill_id,
scope,
workspace,
confirm_update,
} => {
let _installation = self.installation_lock.lock().await;
let catalog = self.catalogs.get(Some(&provider))?;
let skill = {
let _permit = self.provider_permit().await?;
catalog.get(&skill_id).await?
};
let verification_workspace = match scope {
SkillScope::Global => self.runtime_cwd.clone(),
SkillScope::Workspace => {
workspace.as_deref().map(PathBuf::from).ok_or_else(|| {
AgentError::new(ErrorCode::SkillScopeDenied, "workspace denied")
})?
}
};
let pending = self
.store
.begin_install(
&provider,
skill,
scope.clone(),
workspace.as_deref().map(Path::new),
confirm_update,
)
.await?;
let commands = match self.runtime.fresh_commands(&verification_workspace).await {
Ok(commands) => commands,
Err(error) => return rollback_pending(pending, error).await,
};
let expected_name = format!("skill:{}", pending.item().name);
let expected_path = pending.skill_markdown_path().to_path_buf();
let visible = commands.iter().any(|command| {
command.source == "skill"
&& command.name == expected_name
&& command.path.as_deref().is_some_and(|path| {
Path::new(path)
.canonicalize()
.is_ok_and(|path| path == expected_path)
})
});
if !visible {
return rollback_pending(
pending,
AgentError::new(
ErrorCode::SkillInvalid,
"Pi did not discover the installed skill in the target workspace",
),
)
.await;
}
let item = tokio::task::spawn_blocking(move || pending.commit())
.await
.map_err(|_| filesystem_task_failed())??;
Ok(AgentMessage::SkillsInstalled { request_id, item })
}
SkillRequest::Uninstall {
request_id,
installation_id,
} => {
let _installation = self.installation_lock.lock().await;
let store = self.store.clone();
let response_installation_id = installation_id.clone();
tokio::task::spawn_blocking(move || store.uninstall(&installation_id))
.await
.map_err(|_| filesystem_task_failed())??;
Ok(AgentMessage::SkillsUninstalled {
request_id,
installation_id: response_installation_id,
})
}
}
}
async fn provider_permit(&self) -> AgentResult<tokio::sync::OwnedSemaphorePermit> {
self.provider_requests
.clone()
.acquire_owned()
.await
.map_err(|_| {
AgentError::new(
ErrorCode::CatalogUnavailable,
"skill catalog request queue closed",
)
})
}
}
async fn rollback_pending(
pending: crate::infrastructure::skill_store::PendingSkillInstall,
verification_error: AgentError,
) -> AgentResult<AgentMessage> {
tokio::task::spawn_blocking(move || pending.rollback())
.await
.map_err(|_| filesystem_task_failed())??;
Err(verification_error)
}
fn filesystem_task_failed() -> AgentError {
AgentError::new(
ErrorCode::SkillFilesystemFailed,
"skill filesystem task failed",
)
}