shadowdaoandClaude Opus 5.5
f4153dce42
The 21 marketplace commands, registered and granted; marketplace fields kept store-owned in update_settings/update_project; a background refresh of every marketplace at app start. - apply_marketplace_now emits marketplace-sync-finished per project (F4). - Settings export carries marketplace account tokens in ExportedSecrets (account id -> token) and import restores them; imported accounts, marketplaces and global installs are validated with the commands' own rules before anything is written. The import preview discloses the marketplace count, token count and global hook installs, and warns on the latter (F10). - refresh_pins and cache removal hold the repo lock (F11). - ops::validate_host/validate_branch delegate to auth::valid_host and git::valid_branch (F13). - A finished gh container login frees only its own cancel slot. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
776 lines
28 KiB
Rust
776 lines
28 KiB
Rust
//! Marketplaces: git repos of agents, skills, commands, hooks and plugins that
|
|
//! are fetched on the host and synced into containers. See
|
|
//! `docs/superpowers/specs/2026-09-27-marketplace-design.md`.
|
|
|
|
pub mod auth;
|
|
pub mod catalog;
|
|
pub mod diff;
|
|
pub mod gh_login;
|
|
pub mod git;
|
|
pub mod payload;
|
|
pub mod sync;
|
|
pub mod tree;
|
|
#[cfg(test)]
|
|
mod sync_script_tests;
|
|
#[cfg(test)]
|
|
pub(crate) mod test_support;
|
|
|
|
use std::collections::{BTreeSet, HashMap};
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use tauri::Emitter;
|
|
use tokio::sync::oneshot;
|
|
|
|
use crate::models::marketplace::{
|
|
effective_installs, CatalogItem, ItemUpdate, Marketplace, MarketplaceInstall, MarketplaceSnapshot, SyncReport,
|
|
};
|
|
use crate::models::{AppSettings, Project};
|
|
use catalog::{item_fingerprint, parse_catalog};
|
|
use tree::GitTree;
|
|
|
|
/// Emitted after every container sync, payload `{ project_id, report }`.
|
|
pub const SYNC_FINISHED_EVENT: &str = "marketplace-sync-finished";
|
|
|
|
pub struct MarketplaceManager {
|
|
data_root: PathBuf,
|
|
snapshots: Mutex<HashMap<String, MarketplaceSnapshot>>,
|
|
reports: Mutex<HashMap<String, SyncReport>>,
|
|
gh_login_cancel: tokio::sync::Mutex<Option<oneshot::Sender<()>>>,
|
|
/// Serialises writers of the bare caches (fetch, pins, cache removal) so
|
|
/// concurrent refreshes never race on gix ref locks (pre-flight F11a).
|
|
repo_lock: tokio::sync::Mutex<()>,
|
|
/// One lock per project, held for a whole `sync_project`, so a start sync
|
|
/// and Apply now never run `sync.sh` in one container at once (F11b).
|
|
sync_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
|
|
}
|
|
|
|
/// Project ids become file names; anything outside this set is not persisted.
|
|
fn safe_file_stem(id: &str) -> bool {
|
|
!id.is_empty()
|
|
&& id.len() <= 128
|
|
&& id
|
|
.chars()
|
|
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
|
|
}
|
|
|
|
impl MarketplaceManager {
|
|
/// `data_root` is `<data_dir>/triple-c`.
|
|
pub fn new(data_root: PathBuf) -> Self {
|
|
Self {
|
|
data_root,
|
|
snapshots: Mutex::new(HashMap::new()),
|
|
reports: Mutex::new(HashMap::new()),
|
|
gh_login_cancel: tokio::sync::Mutex::new(None),
|
|
repo_lock: tokio::sync::Mutex::new(()),
|
|
sync_locks: Mutex::new(HashMap::new()),
|
|
}
|
|
}
|
|
|
|
pub fn data_root(&self) -> &Path {
|
|
&self.data_root
|
|
}
|
|
|
|
/// Hold while writing to any marketplace cache (fetch, `git::set_pins`,
|
|
/// removing a cache).
|
|
pub fn repo_lock(&self) -> &tokio::sync::Mutex<()> {
|
|
&self.repo_lock
|
|
}
|
|
|
|
/// The project's sync lock; see `sync_project`.
|
|
pub fn sync_lock(&self, project_id: &str) -> Arc<tokio::sync::Mutex<()>> {
|
|
self.sync_locks
|
|
.lock()
|
|
.unwrap()
|
|
.entry(project_id.to_string())
|
|
.or_default()
|
|
.clone()
|
|
}
|
|
|
|
pub fn snapshot(&self, marketplace_id: &str) -> Option<MarketplaceSnapshot> {
|
|
self.snapshots.lock().unwrap().get(marketplace_id).cloned()
|
|
}
|
|
|
|
pub fn put_snapshot(&self, snap: MarketplaceSnapshot) {
|
|
self.snapshots
|
|
.lock()
|
|
.unwrap()
|
|
.insert(snap.marketplace_id.clone(), snap);
|
|
}
|
|
|
|
pub fn remove_snapshot(&self, marketplace_id: &str) {
|
|
self.snapshots.lock().unwrap().remove(marketplace_id);
|
|
}
|
|
|
|
fn report_path(&self, project_id: &str) -> PathBuf {
|
|
self.data_root
|
|
.join("marketplace-sync")
|
|
.join(format!("{project_id}.json"))
|
|
}
|
|
|
|
pub fn report(&self, project_id: &str) -> Option<SyncReport> {
|
|
if let Some(r) = self.reports.lock().unwrap().get(project_id) {
|
|
return Some(r.clone());
|
|
}
|
|
if !safe_file_stem(project_id) {
|
|
return None;
|
|
}
|
|
let text = std::fs::read_to_string(self.report_path(project_id)).ok()?;
|
|
let report: SyncReport = serde_json::from_str(&text).ok()?;
|
|
self.reports
|
|
.lock()
|
|
.unwrap()
|
|
.insert(project_id.to_string(), report.clone());
|
|
Some(report)
|
|
}
|
|
|
|
pub fn put_report(&self, project_id: &str, report: SyncReport) {
|
|
self.reports
|
|
.lock()
|
|
.unwrap()
|
|
.insert(project_id.to_string(), report.clone());
|
|
if !safe_file_stem(project_id) {
|
|
return;
|
|
}
|
|
let path = self.report_path(project_id);
|
|
let write = || -> std::io::Result<()> {
|
|
std::fs::create_dir_all(path.parent().unwrap())?;
|
|
let tmp = path.with_extension("json.tmp");
|
|
std::fs::write(&tmp, serde_json::to_vec_pretty(&report).unwrap_or_default())?;
|
|
std::fs::rename(&tmp, &path)
|
|
};
|
|
if let Err(e) = write() {
|
|
log::warn!(
|
|
"Could not persist the marketplace sync report for {}: {}",
|
|
project_id,
|
|
e
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Claim (`Some`) or release (`None`) the single gh-login slot. Claiming
|
|
/// fails while another login holds it.
|
|
pub async fn set_gh_login_cancel(&self, tx: Option<oneshot::Sender<()>>) -> bool {
|
|
let mut slot = self.gh_login_cancel.lock().await;
|
|
match tx {
|
|
Some(tx) => {
|
|
if slot.is_some() {
|
|
return false;
|
|
}
|
|
*slot = Some(tx);
|
|
true
|
|
}
|
|
None => {
|
|
*slot = None;
|
|
true
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Free the slot after a login ends, but only if it still holds that
|
|
/// login's sender (its receiver is gone once the login returns). A cancel
|
|
/// may have emptied the slot and a newer login claimed it meanwhile; a
|
|
/// plain `set_gh_login_cancel(None)` would drop that login's sender,
|
|
/// which it reads as a cancel.
|
|
pub async fn release_gh_login(&self) {
|
|
let mut slot = self.gh_login_cancel.lock().await;
|
|
if slot.as_ref().is_some_and(|tx| tx.is_closed()) {
|
|
*slot = None;
|
|
}
|
|
}
|
|
|
|
pub async fn cancel_gh_login(&self) {
|
|
if let Some(tx) = self.gh_login_cancel.lock().await.take() {
|
|
let _ = tx.send(());
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Head commit for a marketplace: the in-memory snapshot's, else the cache's.
|
|
pub fn head_for(mgr: &MarketplaceManager, m: &Marketplace) -> Option<String> {
|
|
mgr.snapshot(&m.id).and_then(|s| s.head_commit).or_else(|| {
|
|
git::cached_head(&git::cache_path(mgr.data_root(), &m.id))
|
|
.ok()
|
|
.flatten()
|
|
})
|
|
}
|
|
|
|
fn parse_at(repo: &Path, commit: &str) -> Result<Vec<CatalogItem>, String> {
|
|
let tree = GitTree::open(repo, commit)?;
|
|
Ok(parse_catalog(&tree))
|
|
}
|
|
|
|
/// Snapshot from the cache alone (no network): startup, and after an install
|
|
/// when nothing is in memory. `fetched_at` stays `None`.
|
|
pub fn load_cached_snapshot(
|
|
mgr: &MarketplaceManager,
|
|
marketplace: &Marketplace,
|
|
) -> MarketplaceSnapshot {
|
|
let repo = git::cache_path(mgr.data_root(), &marketplace.id);
|
|
let mut snap = MarketplaceSnapshot {
|
|
marketplace_id: marketplace.id.clone(),
|
|
..Default::default()
|
|
};
|
|
match git::cached_head(&repo) {
|
|
Ok(Some(head)) => match parse_at(&repo, &head) {
|
|
Ok(items) => {
|
|
snap.head_commit = Some(head);
|
|
snap.items = items;
|
|
}
|
|
Err(e) => snap.fetch_error = Some(format!("The cached copy could not be read: {e}")),
|
|
},
|
|
Ok(None) => {}
|
|
Err(e) => snap.fetch_error = Some(format!("The cached copy could not be read: {e}")),
|
|
}
|
|
snap
|
|
}
|
|
|
|
/// Keep the previous items and head (in memory, else from the cache) and
|
|
/// record why this refresh failed.
|
|
fn failed_snapshot(
|
|
mgr: &MarketplaceManager,
|
|
m: &Marketplace,
|
|
message: String,
|
|
) -> MarketplaceSnapshot {
|
|
let mut snap = mgr
|
|
.snapshot(&m.id)
|
|
.unwrap_or_else(|| load_cached_snapshot(mgr, m));
|
|
snap.fetch_error = Some(message);
|
|
mgr.put_snapshot(snap.clone());
|
|
snap
|
|
}
|
|
|
|
/// Refresh one marketplace: resolve the credential, fetch (blocking task, under
|
|
/// the repo lock), parse the catalog at head and store the snapshot. On failure
|
|
/// the previous items and head are kept and `fetch_error` is set.
|
|
pub async fn refresh_marketplace(
|
|
mgr: &MarketplaceManager,
|
|
settings: &AppSettings,
|
|
marketplace_id: &str,
|
|
) -> MarketplaceSnapshot {
|
|
let Some(m) = settings
|
|
.marketplaces
|
|
.iter()
|
|
.find(|m| m.id == marketplace_id)
|
|
.cloned()
|
|
else {
|
|
return MarketplaceSnapshot {
|
|
marketplace_id: marketplace_id.to_string(),
|
|
fetch_error: Some("This marketplace is no longer configured.".to_string()),
|
|
..Default::default()
|
|
};
|
|
};
|
|
let account = m
|
|
.account_id
|
|
.as_ref()
|
|
.and_then(|id| settings.marketplace_accounts.iter().find(|a| &a.id == id))
|
|
.cloned();
|
|
let cred = match &account {
|
|
Some(a) => match auth::resolve_credential(a).await {
|
|
Ok(c) => Some(c),
|
|
Err(e) => return failed_snapshot(mgr, &m, e),
|
|
},
|
|
None => None,
|
|
};
|
|
|
|
let repo = git::cache_path(mgr.data_root(), &m.id);
|
|
let (url, branch) = (m.url.clone(), m.branch.clone());
|
|
let joined = {
|
|
let _repo_guard = mgr.repo_lock.lock().await;
|
|
tokio::task::spawn_blocking(move || {
|
|
let head = git::fetch(&repo, &url, branch.as_deref(), cred)?;
|
|
let items = parse_at(&repo, &head).map_err(git::FetchError::Other)?;
|
|
Ok::<_, git::FetchError>((head, items))
|
|
})
|
|
.await
|
|
};
|
|
|
|
match joined {
|
|
Ok(Ok((head, items))) => {
|
|
let snap = MarketplaceSnapshot {
|
|
marketplace_id: m.id.clone(),
|
|
head_commit: Some(head),
|
|
fetched_at: Some(chrono::Utc::now().to_rfc3339()),
|
|
fetch_error: None,
|
|
items,
|
|
};
|
|
mgr.put_snapshot(snap.clone());
|
|
snap
|
|
}
|
|
Ok(Err(e)) => failed_snapshot(
|
|
mgr,
|
|
&m,
|
|
auth::describe_fetch_error(&e, account.as_ref(), &m.url),
|
|
),
|
|
Err(e) => failed_snapshot(mgr, &m, format!("The refresh task failed: {e}")),
|
|
}
|
|
}
|
|
|
|
fn item_changed(repo: &Path, inst: &MarketplaceInstall, head: &str) -> Result<bool, String> {
|
|
let old = GitTree::open(repo, &inst.commit)?;
|
|
let new = GitTree::open(repo, head)?;
|
|
Ok(item_fingerprint(&old, inst.kind, &inst.key)?
|
|
!= item_fingerprint(&new, inst.kind, &inst.key)?)
|
|
}
|
|
|
|
/// Every install (global + all projects) whose item fingerprint at head
|
|
/// differs from its pin. Installs whose pin is not in the cache are skipped.
|
|
pub fn compute_updates(
|
|
mgr: &MarketplaceManager,
|
|
settings: &AppSettings,
|
|
projects: &[Project],
|
|
) -> Vec<ItemUpdate> {
|
|
let mut seen = BTreeSet::new();
|
|
let mut out = Vec::new();
|
|
let all = settings
|
|
.global_marketplace_installs
|
|
.iter()
|
|
.chain(projects.iter().flat_map(|p| p.marketplace_installs.iter()));
|
|
for inst in all {
|
|
if !seen.insert((inst.item_ref(), inst.commit.clone())) {
|
|
continue;
|
|
}
|
|
let Some(m) = settings
|
|
.marketplaces
|
|
.iter()
|
|
.find(|m| m.id == inst.marketplace_id)
|
|
else {
|
|
continue;
|
|
};
|
|
let Some(head) = head_for(mgr, m) else {
|
|
continue;
|
|
};
|
|
if head == inst.commit {
|
|
continue;
|
|
}
|
|
let repo = git::cache_path(mgr.data_root(), &m.id);
|
|
match item_changed(&repo, inst, &head) {
|
|
Ok(true) => out.push(ItemUpdate {
|
|
item: inst.item_ref(),
|
|
pinned: inst.commit.clone(),
|
|
head,
|
|
}),
|
|
Ok(false) => {}
|
|
Err(e) => log::debug!("Update check skipped for {}: {}", inst.key, e),
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
fn project_installs(settings: &AppSettings, project: &Project) -> Vec<MarketplaceInstall> {
|
|
effective_installs(
|
|
&settings.global_marketplace_installs,
|
|
&project.marketplace_disabled,
|
|
&project.marketplace_installs,
|
|
)
|
|
}
|
|
|
|
/// Build the project's payload and sync it into its running container. The
|
|
/// report is stored (and persisted) whatever happens. Holds the project's sync
|
|
/// lock throughout, so concurrent syncs of one project run one after another.
|
|
pub async fn sync_project(
|
|
mgr: &MarketplaceManager,
|
|
settings: &AppSettings,
|
|
project: &Project,
|
|
container_id: &str,
|
|
) -> SyncReport {
|
|
let lock = mgr.sync_lock(&project.id);
|
|
let _sync_guard = lock.lock().await;
|
|
|
|
let installs = project_installs(settings, project);
|
|
let marketplaces = settings.marketplaces.clone();
|
|
let root = mgr.data_root().to_path_buf();
|
|
let built = tokio::task::spawn_blocking(move || {
|
|
payload::build_payload(&payload::PayloadInput {
|
|
installs: &installs,
|
|
marketplaces: &marketplaces,
|
|
data_root: &root,
|
|
})
|
|
})
|
|
.await
|
|
.map_err(|e| format!("Building the marketplace payload failed: {e}"))
|
|
.and_then(|r| r);
|
|
|
|
let report = match built {
|
|
Ok(p) => {
|
|
let items = p.manifest["items"].as_array().map_or(0, Vec::len);
|
|
log::debug!(
|
|
"Marketplace sync for project {}: {} item(s) in the payload, {} skipped on the host",
|
|
project.id,
|
|
items,
|
|
p.skipped.len()
|
|
);
|
|
let result = sync::sync_container(container_id, &p).await;
|
|
sync::with_payload_skips(sync::report_from_result(result), &p.skipped)
|
|
}
|
|
Err(e) => sync::report_from_result(Err(e)),
|
|
};
|
|
mgr.put_report(&project.id, report.clone());
|
|
report
|
|
}
|
|
|
|
/// A project with no items that has never been synced has nothing to add and
|
|
/// nothing to remove, so its start does not wait on a sync at all.
|
|
pub fn should_sync(mgr: &MarketplaceManager, settings: &AppSettings, project: &Project) -> bool {
|
|
!project_installs(settings, project).is_empty() || mgr.report(&project.id).is_some()
|
|
}
|
|
|
|
/// Sync in the background after a container start. The sync waits for the
|
|
/// entrypoint to finish (which can include a two-minute `claude update`), and
|
|
/// its failure must never fail the start — so the start never awaits it.
|
|
pub fn spawn_project_sync(
|
|
app: tauri::AppHandle,
|
|
mgr: Arc<MarketplaceManager>,
|
|
settings: AppSettings,
|
|
project: Project,
|
|
container_id: String,
|
|
) {
|
|
if !should_sync(&mgr, &settings, &project) {
|
|
return;
|
|
}
|
|
tauri::async_runtime::spawn(async move {
|
|
let report = sync_project(&mgr, &settings, &project, &container_id).await;
|
|
if !report.errors.is_empty() {
|
|
log::warn!(
|
|
"Marketplace sync for project {} reported errors: {:?}",
|
|
project.id,
|
|
report.errors
|
|
);
|
|
}
|
|
let _ = app.emit(
|
|
SYNC_FINISHED_EVENT,
|
|
serde_json::json!({ "project_id": project.id, "report": report }),
|
|
);
|
|
});
|
|
}
|
|
|
|
/// All commits referenced by installs, per marketplace (for `git::set_pins`).
|
|
pub fn pins_by_marketplace(
|
|
settings: &AppSettings,
|
|
projects: &[Project],
|
|
) -> HashMap<String, Vec<String>> {
|
|
let mut map: HashMap<String, BTreeSet<String>> = HashMap::new();
|
|
let all = settings
|
|
.global_marketplace_installs
|
|
.iter()
|
|
.chain(projects.iter().flat_map(|p| p.marketplace_installs.iter()));
|
|
for inst in all {
|
|
map.entry(inst.marketplace_id.clone())
|
|
.or_default()
|
|
.insert(inst.commit.clone());
|
|
}
|
|
map.into_iter()
|
|
.map(|(k, v)| (k, v.into_iter().collect()))
|
|
.collect()
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::marketplace::test_support::GitFixture;
|
|
use crate::models::marketplace::{ItemKind, Marketplace, MarketplaceInstall};
|
|
|
|
fn settings_with(url: &str) -> AppSettings {
|
|
let mut s = AppSettings::default();
|
|
s.marketplaces.push(Marketplace {
|
|
id: "m1".into(),
|
|
name: "Test".into(),
|
|
url: url.into(),
|
|
branch: None,
|
|
account_id: None,
|
|
});
|
|
s
|
|
}
|
|
|
|
fn install(kind: ItemKind, key: &str, commit: &str) -> MarketplaceInstall {
|
|
MarketplaceInstall {
|
|
marketplace_id: "m1".into(),
|
|
kind,
|
|
key: key.into(),
|
|
commit: commit.into(),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn refresh_parses_the_catalog_at_head() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
|
|
let snap = refresh_marketplace(&mgr, &settings_with(&fx.url()), "m1").await;
|
|
|
|
assert_eq!(snap.fetch_error, None);
|
|
assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
|
|
assert!(snap.fetched_at.is_some());
|
|
let mut keys: Vec<String> = snap
|
|
.items
|
|
.iter()
|
|
.map(|i| format!("{:?}:{}", i.kind, i.key))
|
|
.collect();
|
|
keys.sort();
|
|
assert_eq!(
|
|
keys,
|
|
vec![
|
|
"Agent:code-reviewer",
|
|
"Command:example-command",
|
|
"Hook:notify-on-stop",
|
|
"Plugin:example-plugin",
|
|
"Skill:example-skill",
|
|
]
|
|
);
|
|
assert_eq!(mgr.snapshot("m1"), Some(snap));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn refresh_failure_keeps_snapshot() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let url = fx.url();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let settings = settings_with(&url);
|
|
let first = refresh_marketplace(&mgr, &settings, "m1").await;
|
|
assert_eq!(first.fetch_error, None);
|
|
|
|
drop(fx); // the source repository disappears (offline, deleted, …)
|
|
let second = refresh_marketplace(&mgr, &settings, "m1").await;
|
|
|
|
assert!(second.fetch_error.is_some(), "expected a fetch error");
|
|
assert_eq!(second.head_commit.as_deref(), Some(c1.as_str()));
|
|
assert_eq!(second.items, first.items);
|
|
assert_eq!(second.fetched_at, first.fetched_at);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn refresh_waits_for_the_repo_lock() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let settings = settings_with(&fx.url());
|
|
|
|
let guard = mgr.repo_lock().lock().await;
|
|
let blocked = tokio::time::timeout(
|
|
std::time::Duration::from_millis(300),
|
|
refresh_marketplace(&mgr, &settings, "m1"),
|
|
)
|
|
.await;
|
|
assert!(blocked.is_err(), "refresh must not fetch while the repo lock is held");
|
|
assert!(
|
|
!git::cache_path(data.path(), "m1").exists(),
|
|
"nothing may touch the cache while the lock is held"
|
|
);
|
|
drop(guard);
|
|
|
|
let snap = refresh_marketplace(&mgr, &settings, "m1").await;
|
|
assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn concurrent_refreshes_all_succeed() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let settings = settings_with(&fx.url());
|
|
|
|
let (a, b, c) = tokio::join!(
|
|
refresh_marketplace(&mgr, &settings, "m1"),
|
|
refresh_marketplace(&mgr, &settings, "m1"),
|
|
refresh_marketplace(&mgr, &settings, "m1"),
|
|
);
|
|
for snap in [a, b, c] {
|
|
assert_eq!(snap.fetch_error, None);
|
|
assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cached_snapshot_loads_without_network() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
let data = tempfile::tempdir().unwrap();
|
|
let settings = settings_with(&fx.url());
|
|
{
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
refresh_marketplace(&mgr, &settings, "m1").await;
|
|
}
|
|
drop(fx);
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let snap = load_cached_snapshot(&mgr, &settings.marketplaces[0]);
|
|
assert_eq!(snap.head_commit.as_deref(), Some(c1.as_str()));
|
|
assert_eq!(snap.items.len(), 5);
|
|
assert_eq!(snap.fetch_error, None);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn only_items_whose_own_files_changed_have_updates() {
|
|
let Some(fx) = GitFixture::new() else { return };
|
|
let c1 = fx.with_all_kinds();
|
|
fx.write(
|
|
"agents/code-reviewer.md",
|
|
"---\nname: code-reviewer\ndescription: Reviews code\n---\nReview harder.\n",
|
|
);
|
|
let c2 = fx.commit("tweak agent");
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let mut settings = settings_with(&fx.url());
|
|
settings.global_marketplace_installs = vec![
|
|
install(ItemKind::Agent, "code-reviewer", &c1),
|
|
install(ItemKind::Hook, "notify-on-stop", &c1),
|
|
];
|
|
let mut project = crate::models::Project::new("p".into(), vec![]);
|
|
project.marketplace_installs = vec![install(ItemKind::Skill, "example-skill", &c1)];
|
|
refresh_marketplace(&mgr, &settings, "m1").await;
|
|
|
|
let updates = compute_updates(&mgr, &settings, &[project]);
|
|
|
|
assert_eq!(updates.len(), 1, "{updates:?}");
|
|
assert_eq!(updates[0].item.key, "code-reviewer");
|
|
assert_eq!(updates[0].pinned, c1);
|
|
assert_eq!(updates[0].head, c2);
|
|
}
|
|
|
|
#[test]
|
|
fn pins_are_grouped_and_deduplicated_per_marketplace() {
|
|
let a = "a".repeat(40);
|
|
let b = "b".repeat(40);
|
|
let mut settings = settings_with("https://example.invalid/r.git");
|
|
settings.global_marketplace_installs = vec![
|
|
install(ItemKind::Agent, "x", &b),
|
|
install(ItemKind::Hook, "y", &a),
|
|
];
|
|
let mut project = crate::models::Project::new("p".into(), vec![]);
|
|
project.marketplace_installs = vec![install(ItemKind::Agent, "z", &a)];
|
|
let pins = pins_by_marketplace(&settings, &[project]);
|
|
assert_eq!(pins.get("m1"), Some(&vec![a.clone(), b.clone()]));
|
|
}
|
|
|
|
#[test]
|
|
fn reports_are_persisted_per_project() {
|
|
let data = tempfile::tempdir().unwrap();
|
|
let report = SyncReport {
|
|
installed: vec!["agent:x".into()],
|
|
..Default::default()
|
|
};
|
|
MarketplaceManager::new(data.path().to_path_buf()).put_report("proj-1", report.clone());
|
|
let fresh = MarketplaceManager::new(data.path().to_path_buf());
|
|
assert_eq!(fresh.report("proj-1"), Some(report));
|
|
assert_eq!(fresh.report("proj-2"), None);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn only_one_gh_login_may_hold_the_cancel_slot() {
|
|
let mgr = MarketplaceManager::new(std::env::temp_dir());
|
|
let (tx1, rx1) = tokio::sync::oneshot::channel();
|
|
let (tx2, _rx2) = tokio::sync::oneshot::channel();
|
|
assert!(mgr.set_gh_login_cancel(Some(tx1)).await);
|
|
assert!(!mgr.set_gh_login_cancel(Some(tx2)).await);
|
|
mgr.cancel_gh_login().await;
|
|
assert!(rx1.await.is_ok(), "cancel must signal the running login");
|
|
let (tx3, _rx3) = tokio::sync::oneshot::channel();
|
|
assert!(
|
|
mgr.set_gh_login_cancel(Some(tx3)).await,
|
|
"slot is free after cancel"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn releasing_a_finished_login_never_frees_a_newer_ones_slot() {
|
|
let mgr = MarketplaceManager::new(std::env::temp_dir());
|
|
// Login A is cancelled, and login B claims the slot before A returns.
|
|
let (tx_a, rx_a) = tokio::sync::oneshot::channel::<()>();
|
|
assert!(mgr.set_gh_login_cancel(Some(tx_a)).await);
|
|
mgr.cancel_gh_login().await;
|
|
let (tx_b, mut rx_b) = tokio::sync::oneshot::channel::<()>();
|
|
assert!(mgr.set_gh_login_cancel(Some(tx_b)).await);
|
|
drop(rx_a); // A returns.
|
|
mgr.release_gh_login().await;
|
|
assert!(
|
|
matches!(rx_b.try_recv(), Err(tokio::sync::oneshot::error::TryRecvError::Empty)),
|
|
"B's sender must still be held, not dropped"
|
|
);
|
|
// B returns: its slot is freed.
|
|
drop(rx_b);
|
|
mgr.release_gh_login().await;
|
|
let (tx_c, _rx_c) = tokio::sync::oneshot::channel::<()>();
|
|
assert!(mgr.set_gh_login_cancel(Some(tx_c)).await);
|
|
}
|
|
|
|
#[test]
|
|
fn a_project_that_never_had_items_is_not_synced() {
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let settings = settings_with("https://example.invalid/r.git");
|
|
let project = crate::models::Project::new("p".into(), vec![]);
|
|
assert!(!should_sync(&mgr, &settings, &project));
|
|
}
|
|
|
|
#[test]
|
|
fn a_project_with_items_or_a_previous_sync_is_synced() {
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let mut settings = settings_with("https://example.invalid/r.git");
|
|
let project = crate::models::Project::new("p".into(), vec![]);
|
|
settings.global_marketplace_installs = vec![install(ItemKind::Agent, "a", &"a".repeat(40))];
|
|
assert!(should_sync(&mgr, &settings, &project), "global items apply");
|
|
|
|
// Everything was uninstalled since the last sync: the container still
|
|
// holds the old files, so it must be synced to remove them.
|
|
settings.global_marketplace_installs.clear();
|
|
mgr.put_report(&project.id, SyncReport::default());
|
|
assert!(should_sync(&mgr, &settings, &project));
|
|
}
|
|
|
|
#[test]
|
|
fn a_project_whose_only_item_is_disabled_is_not_synced() {
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let mut settings = settings_with("https://example.invalid/r.git");
|
|
let inst = install(ItemKind::Agent, "a", &"a".repeat(40));
|
|
let mut project = crate::models::Project::new("p".into(), vec![]);
|
|
project.marketplace_disabled = vec![inst.item_ref()];
|
|
settings.global_marketplace_installs = vec![inst];
|
|
assert!(!should_sync(&mgr, &settings, &project));
|
|
}
|
|
|
|
#[test]
|
|
fn sync_locks_are_per_project() {
|
|
let mgr = MarketplaceManager::new(std::env::temp_dir());
|
|
let a1 = mgr.sync_lock("a");
|
|
let a2 = mgr.sync_lock("a");
|
|
let b = mgr.sync_lock("b");
|
|
assert!(Arc::ptr_eq(&a1, &a2), "one lock per project");
|
|
assert!(!Arc::ptr_eq(&a1, &b), "projects do not block each other");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn sync_project_waits_for_the_projects_sync_lock() {
|
|
// Pre-flight F11b: a start sync and Apply now must never run sync.sh
|
|
// in the same container at once.
|
|
let data = tempfile::tempdir().unwrap();
|
|
let mgr = MarketplaceManager::new(data.path().to_path_buf());
|
|
let settings = settings_with("https://example.invalid/r.git");
|
|
let project = crate::models::Project::new("p".into(), vec![]);
|
|
|
|
let lock = mgr.sync_lock(&project.id);
|
|
let guard = lock.lock().await;
|
|
let blocked = tokio::time::timeout(
|
|
std::time::Duration::from_millis(300),
|
|
sync_project(&mgr, &settings, &project, "no-such-container"),
|
|
)
|
|
.await;
|
|
assert!(blocked.is_err(), "sync must wait while another sync holds the lock");
|
|
assert_eq!(mgr.report(&project.id), None, "nothing ran while blocked");
|
|
drop(guard);
|
|
|
|
// No Docker (or no such container) here: the failure becomes a stored
|
|
// report instead of an error.
|
|
let report = sync_project(&mgr, &settings, &project, "no-such-container").await;
|
|
assert_eq!(report.errors.len(), 1, "{report:?}");
|
|
assert!(!report.finished_at.is_empty());
|
|
assert_eq!(mgr.report(&project.id), Some(report));
|
|
}
|
|
}
|