use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use futures_util::{Stream, StreamExt};
use reqwest::{Client, Response, StatusCode, Url, redirect::Policy};
use serde::Deserialize;
use serde::de::DeserializeOwned;
use crate::config::Config;
use crate::domain::errors::{AgentError, AgentResult, ErrorCode};
use crate::domain::protocol::SkillSearchItem;
use crate::domain::skills::{
CatalogSearch, CatalogSkill, MAX_RESPONSE_BYTES, MAX_SEARCH_LIMIT, SkillCatalog, SkillFile,
SkillSource,
};
const SKILLSPOOL_ID: &str = "skillspool";
const SKILLS_SH_ID: &str = "skills_sh";
const SKILLSPOOL_BASE_URL: &str = "https://skillspool.org";
const SKILLS_SH_BASE_URL: &str = "https://skills.sh";
pub(crate) fn build_http_client() -> AgentResult<Client> {
Client::builder()
.timeout(Duration::from_secs(15))
.redirect(Policy::limited(5))
.user_agent(concat!("regy-pc-agent/", env!("CARGO_PKG_VERSION")))
.no_proxy()
.build()
.map_err(|error| {
AgentError::new(
ErrorCode::CatalogNotConfigured,
format!("failed to build catalog HTTP client: {error}"),
)
})
}
#[derive(Debug, Clone)]
pub struct SkillspoolCatalog {
base_url: Url,
client: Client,
}
impl SkillspoolCatalog {
pub fn new(base_url: String, client: Client) -> AgentResult<Self> {
Ok(Self {
base_url: parse_base_url(&base_url)?,
client,
})
}
fn endpoint(&self, segments: &[&str]) -> AgentResult<Url> {
endpoint(&self.base_url, segments)
}
}
#[async_trait]
impl SkillCatalog for SkillspoolCatalog {
fn id(&self) -> &'static str {
SKILLSPOOL_ID
}
async fn search(&self, query: &str, limit: u16, offset: u32) -> AgentResult<CatalogSearch> {
validate_limit(limit)?;
let searching = !query.trim().is_empty();
let segments: &[&str] = if searching {
&["api", "v1", "skills", "search"]
} else {
&["api", "v1", "skills"]
};
let mut url = self.endpoint(segments)?;
{
let mut pairs = url.query_pairs_mut();
if searching {
pairs.append_pair("q", query);
}
pairs
.append_pair("limit", &limit.to_string())
.append_pair("offset", &offset.to_string());
}
let response: SkillspoolSearchResponse =
send_json(self.client.get(url).send().await).await?;
let item_count = response.data.len() as u32;
let has_more = response
.total
.map_or(item_count >= u32::from(limit), |total| {
offset.saturating_add(item_count) < total
});
let next_offset = has_more
.then(|| offset.checked_add(u32::from(limit)))
.flatten();
let items = response
.data
.into_iter()
.map(|item| item.normalize(&self.base_url))
.collect::<AgentResult<_>>()?;
Ok(CatalogSearch { items, next_offset })
}
async fn get(&self, skill_id: &str) -> AgentResult<CatalogSkill> {
let url = self.endpoint(&["api", "v1", "skills", skill_id])?;
let response: SkillspoolDetailResponse =
send_json(self.client.get(url).send().await).await?;
let detail = response.data;
let repository = parse_github_tree_url(&detail.code_repository)?;
let item = SkillSearchItem {
id: detail.slug.clone(),
name: detail.name,
description: detail.description,
author: detail.author,
source_url: self
.endpoint(&["api", "v1", "skills", &detail.slug])?
.to_string(),
repository_url: Some(repository.repository.clone()),
popularity: detail.stars,
security_status: None,
updated_at: detail.date_modified,
};
Ok(CatalogSkill {
item,
source: SkillSource {
repository: repository.repository,
revision: repository.revision,
directory: repository.directory,
files: None,
content_hash: None,
},
})
}
}
#[derive(Clone)]
pub struct SkillsShCatalog {
base_url: Url,
token: Arc<str>,
client: Client,
}
impl std::fmt::Debug for SkillsShCatalog {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("SkillsShCatalog")
.field("base_url", &self.base_url)
.field("token", &"<redacted>")
.finish_non_exhaustive()
}
}
impl SkillsShCatalog {
pub fn new(base_url: String, token: String, client: Client) -> AgentResult<Self> {
if token.trim().is_empty() {
return Err(AgentError::new(
ErrorCode::CatalogNotConfigured,
"skills_sh requires PC_AGENT_SKILLS_API_TOKEN",
));
}
Ok(Self {
base_url: parse_base_url(&base_url)?,
token: token.into(),
client,
})
}
fn endpoint(&self, segments: &[&str]) -> AgentResult<Url> {
endpoint(&self.base_url, segments)
}
fn get_request(&self, url: Url) -> reqwest::RequestBuilder {
self.client.get(url).bearer_auth(self.token.as_ref())
}
}
#[async_trait]
impl SkillCatalog for SkillsShCatalog {
fn id(&self) -> &'static str {
SKILLS_SH_ID
}
async fn search(&self, query: &str, limit: u16, offset: u32) -> AgentResult<CatalogSearch> {
validate_limit(limit)?;
let searching = !query.trim().is_empty();
if searching && offset > 0 {
return Ok(CatalogSearch {
items: Vec::new(),
next_offset: None,
});
}
if searching {
let mut url = self.endpoint(&["api", "v1", "skills", "search"])?;
url.query_pairs_mut()
.append_pair("q", query)
.append_pair("limit", &limit.to_string());
let response: SkillsShSearchResponse =
send_json(self.get_request(url).send().await).await?;
let items = normalize_skills_sh_items(response.data);
return Ok(CatalogSearch {
items,
next_offset: None,
});
}
let mut url = self.endpoint(&["api", "v1", "skills"])?;
url.query_pairs_mut()
.append_pair("view", "all-time")
.append_pair("page", &(offset / u32::from(limit)).to_string())
.append_pair("per_page", &limit.to_string());
let response: SkillsShBrowseResponse =
send_json(self.get_request(url).send().await).await?;
let next_offset = response
.pagination
.has_more
.then(|| offset.saturating_add(limit as u32));
let items = normalize_skills_sh_items(response.data);
Ok(CatalogSearch { items, next_offset })
}
async fn get(&self, skill_id: &str) -> AgentResult<CatalogSkill> {
let mut segments = vec!["api", "v1", "skills"];
segments.extend(skill_id.split('/'));
let response: SkillsShDetail =
send_json(self.get_request(self.endpoint(&segments)?).send().await).await?;
let repository_url = github_repository_from_source(&response.source);
let repository = repository_url
.clone()
.unwrap_or_else(|| response.source.clone());
let author = source_author(&response.source);
let mut source_segments = vec!["api", "v1", "skills"];
source_segments.extend(response.id.split('/'));
let source_url = self.endpoint(&source_segments)?.to_string();
let files = response.files.map(|files| {
files
.into_iter()
.map(|file| SkillFile::text(file.path, file.contents))
.collect()
});
let item = SkillSearchItem {
id: response.id,
name: response.slug.clone(),
description: String::new(),
author,
source_url,
repository_url,
popularity: response.installs,
security_status: None,
updated_at: None,
};
Ok(CatalogSkill {
item,
source: SkillSource {
repository,
revision: response.hash.clone().unwrap_or_default(),
directory: response.slug,
files,
content_hash: response.hash,
},
})
}
}
pub struct SkillCatalogRegistry {
default_provider: String,
providers: HashMap<String, Arc<dyn SkillCatalog>>,
}
impl std::fmt::Debug for SkillCatalogRegistry {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let mut providers: Vec<&str> = self.providers.keys().map(String::as_str).collect();
providers.sort_unstable();
formatter
.debug_struct("SkillCatalogRegistry")
.field("default_provider", &self.default_provider)
.field("providers", &providers)
.finish()
}
}
impl SkillCatalogRegistry {
#[cfg(test)]
pub(crate) fn from_test_catalog(catalog: Arc<dyn SkillCatalog>) -> Self {
let default_provider = catalog.id().to_string();
let providers = HashMap::from([(default_provider.clone(), catalog)]);
Self {
default_provider,
providers,
}
}
pub fn from_config(config: &Config) -> AgentResult<Self> {
if !matches!(
config.skills_provider.as_str(),
SKILLSPOOL_ID | SKILLS_SH_ID
) {
return Err(catalog_not_configured(&config.skills_provider));
}
let client = build_http_client()?;
let mut providers: HashMap<String, Arc<dyn SkillCatalog>> = HashMap::new();
let skillspool_base = if config.skills_provider == SKILLSPOOL_ID {
config.skills_api_base_url.clone()
} else {
SKILLSPOOL_BASE_URL.to_string()
};
providers.insert(
SKILLSPOOL_ID.into(),
Arc::new(SkillspoolCatalog::new(skillspool_base, client.clone())?),
);
if let Some(token) = &config.skills_api_token {
let skills_sh_base = if config.skills_provider == SKILLS_SH_ID {
config.skills_api_base_url.clone()
} else {
SKILLS_SH_BASE_URL.to_string()
};
providers.insert(
SKILLS_SH_ID.into(),
Arc::new(SkillsShCatalog::new(skills_sh_base, token.clone(), client)?),
);
}
if !providers.contains_key(&config.skills_provider) {
return Err(catalog_not_configured(&config.skills_provider));
}
Ok(Self {
default_provider: config.skills_provider.clone(),
providers,
})
}
pub fn get(&self, requested: Option<&str>) -> AgentResult<Arc<dyn SkillCatalog>> {
let provider = requested.unwrap_or(&self.default_provider);
self.providers
.get(provider)
.cloned()
.ok_or_else(|| catalog_not_configured(provider))
}
}
fn parse_base_url(base_url: &str) -> AgentResult<Url> {
let mut url = Url::parse(base_url).map_err(|error| {
AgentError::new(
ErrorCode::CatalogNotConfigured,
format!("invalid catalog base URL: {error}"),
)
})?;
if !matches!(url.scheme(), "http" | "https") || url.cannot_be_a_base() {
return Err(AgentError::new(
ErrorCode::CatalogNotConfigured,
"catalog base URL must use http or https",
));
}
url.set_query(None);
url.set_fragment(None);
Ok(url)
}
fn endpoint(base_url: &Url, segments: &[&str]) -> AgentResult<Url> {
let mut url = base_url.clone();
{
let mut path = url.path_segments_mut().map_err(|_| {
AgentError::new(
ErrorCode::CatalogNotConfigured,
"catalog base URL cannot contain path segments",
)
})?;
path.pop_if_empty();
path.extend(segments);
}
Ok(url)
}
fn validate_limit(limit: u16) -> AgentResult<()> {
if limit == 0 || limit > MAX_SEARCH_LIMIT {
return Err(AgentError::new(
ErrorCode::InvalidMessage,
format!("skill search limit must be between 1 and {MAX_SEARCH_LIMIT}"),
));
}
Ok(())
}
async fn send_json<T>(response: Result<Response, reqwest::Error>) -> AgentResult<T>
where
T: DeserializeOwned,
{
let response = response.map_err(|error| {
AgentError::new(
ErrorCode::CatalogUnavailable,
format!("catalog request failed: {error}"),
)
})?;
map_status(response.status())?;
if response
.content_length()
.is_some_and(|size| size > MAX_RESPONSE_BYTES as u64)
{
return Err(invalid_response("catalog response exceeds size limit"));
}
let body = read_bounded_stream(response.bytes_stream(), MAX_RESPONSE_BYTES).await?;
serde_json::from_slice(&body)
.map_err(|error| invalid_response(format!("invalid catalog response: {error}")))
}
pub(crate) async fn read_bounded_stream<S, B, E>(
stream: S,
max_bytes: usize,
) -> AgentResult<Vec<u8>>
where
S: Stream<Item = Result<B, E>>,
B: AsRef<[u8]>,
E: std::fmt::Display,
{
futures_util::pin_mut!(stream);
let mut body = Vec::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|error| {
AgentError::new(
ErrorCode::CatalogUnavailable,
format!("failed to read catalog response: {error}"),
)
})?;
let bytes = chunk.as_ref();
if bytes.len() > max_bytes.saturating_sub(body.len()) {
return Err(invalid_response("catalog response exceeds size limit"));
}
body.extend_from_slice(bytes);
}
Ok(body)
}
fn map_status(status: StatusCode) -> AgentResult<()> {
let code = match status.as_u16() {
401 | 403 => ErrorCode::CatalogAuthenticationFailed,
404 => ErrorCode::SkillNotFound,
429 => ErrorCode::CatalogRateLimited,
500..=599 => ErrorCode::CatalogUnavailable,
_ if !status.is_success() => ErrorCode::CatalogUnavailable,
_ => return Ok(()),
};
Err(AgentError::new(
code,
format!("catalog request returned HTTP {status}"),
))
}
fn invalid_response(message: impl Into<String>) -> AgentError {
AgentError::new(ErrorCode::CatalogResponseInvalid, message)
}
fn catalog_not_configured(provider: &str) -> AgentError {
AgentError::new(
ErrorCode::CatalogNotConfigured,
format!("catalog provider {provider:?} is not configured"),
)
}
#[derive(Debug, Deserialize)]
struct SkillspoolSearchResponse {
data: Vec<SkillspoolItem>,
total: Option<u32>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct SkillspoolItem {
slug: String,
name: String,
description: String,
author: Option<String>,
stars: Option<u64>,
date_modified: Option<String>,
}
impl SkillspoolItem {
fn normalize(self, base_url: &Url) -> AgentResult<SkillSearchItem> {
Ok(SkillSearchItem {
source_url: endpoint(base_url, &["api", "v1", "skills", &self.slug])?.to_string(),
id: self.slug,
name: self.name,
description: self.description,
author: self.author,
repository_url: None,
popularity: self.stars,
security_status: None,
updated_at: self.date_modified,
})
}
}
#[derive(Debug, Deserialize)]
struct SkillspoolDetailResponse {
data: SkillspoolDetail,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct SkillspoolDetail {
slug: String,
name: String,
description: String,
author: Option<String>,
stars: Option<u64>,
code_repository: String,
date_modified: Option<String>,
#[serde(rename = "content")]
_content: Option<String>,
}
struct GithubTree {
repository: String,
revision: String,
directory: String,
}
fn parse_github_tree_url(raw: &str) -> AgentResult<GithubTree> {
let url = Url::parse(raw)
.map_err(|error| invalid_response(format!("invalid codeRepository URL: {error}")))?;
if url.scheme() != "https" || url.host_str() != Some("github.com") {
return Err(invalid_response(
"codeRepository must be an https://github.com URL",
));
}
let segments: Vec<_> = url
.path_segments()
.into_iter()
.flatten()
.filter(|segment| !segment.is_empty())
.collect();
if segments.len() < 5 || segments[2] != "tree" {
return Err(invalid_response(
"codeRepository must identify a repository revision and directory",
));
}
Ok(GithubTree {
repository: format!("https://github.com/{}/{}", segments[0], segments[1]),
revision: segments[3].to_string(),
directory: segments[4..].join("/"),
})
}
#[derive(Debug, Deserialize)]
struct SkillsShBrowseResponse {
data: Vec<SkillsShItem>,
pagination: SkillsShPagination,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct SkillsShPagination {
has_more: bool,
}
#[derive(Debug, Deserialize)]
struct SkillsShSearchResponse {
data: Vec<SkillsShItem>,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct SkillsShItem {
id: String,
name: String,
source: String,
installs: Option<u64>,
source_type: String,
install_url: Option<String>,
url: String,
}
fn normalize_skills_sh_items(items: Vec<SkillsShItem>) -> Vec<SkillSearchItem> {
items
.into_iter()
.map(|item| {
let repository_url = if item.source_type == "well-known" {
None
} else {
item.install_url
.as_deref()
.and_then(github_repository_from_url)
.or_else(|| github_repository_from_source(&item.source))
};
SkillSearchItem {
id: item.id,
name: item.name,
description: String::new(),
author: source_author(&item.source),
source_url: item.url,
repository_url,
popularity: item.installs,
security_status: None,
updated_at: None,
}
})
.collect()
}
fn github_repository_from_source(source: &str) -> Option<String> {
let segments: Vec<_> = source
.split('/')
.filter(|segment| !segment.is_empty())
.collect();
if segments.len() != 2
|| segments
.iter()
.any(|segment| *segment == "." || *segment == "..")
{
return None;
}
Some(format!(
"https://github.com/{}/{}",
segments[0], segments[1]
))
}
fn github_repository_from_url(raw: &str) -> Option<String> {
let url = Url::parse(raw).ok()?;
if url.scheme() != "https" || url.host_str() != Some("github.com") {
return None;
}
let mut segments = url.path_segments()?.filter(|segment| !segment.is_empty());
let owner = segments.next()?;
let repository = segments.next()?;
Some(format!("https://github.com/{owner}/{repository}"))
}
fn source_author(source: &str) -> Option<String> {
source
.split('/')
.find(|segment| !segment.is_empty())
.map(str::to_string)
}
#[derive(Debug, Deserialize)]
struct SkillsShDetail {
id: String,
source: String,
slug: String,
installs: Option<u64>,
hash: Option<String>,
files: Option<Vec<SkillsShFile>>,
}
#[derive(Debug, Deserialize)]
struct SkillsShFile {
path: String,
contents: String,
}