Triple-C marketplace: agents, skills, commands, hooks and plugins for all or individual containers #65
@@ -1449,6 +1449,16 @@ async fn start_project_container_locked(
|
||||
log::warn!("Failed to sync AWS credentials for project {}: {}", project.id, e);
|
||||
}
|
||||
|
||||
// Marketplace items sync in the background — see `spawn_project_sync`
|
||||
// for why the start never waits on it or fails because of it.
|
||||
crate::marketplace::spawn_project_sync(
|
||||
app_handle.clone(),
|
||||
state.marketplace.clone(),
|
||||
state.settings_store.get(),
|
||||
project.clone(),
|
||||
container_id.clone(),
|
||||
);
|
||||
|
||||
Ok(container_id)
|
||||
}.await;
|
||||
|
||||
|
||||
@@ -434,7 +434,8 @@ fn container_join(dir: &str, name: &str) -> String {
|
||||
/// Write `data` into the container at `<dest_dir>/<file_name>` with `mode`.
|
||||
///
|
||||
/// For small, generated files — migration uses it for the `tar -T` include
|
||||
/// list, which can be too long to pass as argv. Anything large should be
|
||||
/// list, which can be too long to pass as argv, and the marketplace sync for
|
||||
/// its payload tar and script. Anything large should be
|
||||
/// streamed through an attached exec's stdin instead, since this buffers the
|
||||
/// whole payload in memory twice (once raw, once tarred).
|
||||
pub async fn upload_bytes_to_container(
|
||||
@@ -446,9 +447,10 @@ pub async fn upload_bytes_to_container(
|
||||
) -> Result<String, String> {
|
||||
let docker = get_docker()?;
|
||||
|
||||
// Root-owned on purpose: the only caller is migration, whose `tar -T` list
|
||||
// is read back as root. The mtime still gets stamped so the file doesn't
|
||||
// read as 1970.
|
||||
// Root-owned on purpose: migration's `tar -T` list is read back as root,
|
||||
// and the marketplace sync uploads into a `claude`-owned directory it
|
||||
// prepares first, so `claude` can still read and delete the files. The
|
||||
// mtime still gets stamped so the file doesn't read as 1970.
|
||||
let tar_buf = build_single_file_tar(file_name, data, mode, 0, 0, now_epoch_secs())?;
|
||||
|
||||
docker
|
||||
|
||||
@@ -16,12 +16,13 @@ pub(crate) mod test_support;
|
||||
|
||||
use std::collections::{BTreeSet, HashMap};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Mutex;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use tauri::Emitter;
|
||||
use tokio::sync::oneshot;
|
||||
|
||||
use crate::models::marketplace::{
|
||||
CatalogItem, ItemUpdate, Marketplace, MarketplaceInstall, MarketplaceSnapshot, SyncReport,
|
||||
effective_installs, CatalogItem, ItemUpdate, Marketplace, MarketplaceInstall, MarketplaceSnapshot, SyncReport,
|
||||
};
|
||||
use crate::models::{AppSettings, Project};
|
||||
use catalog::{item_fingerprint, parse_catalog};
|
||||
@@ -38,6 +39,9 @@ pub struct MarketplaceManager {
|
||||
/// 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.
|
||||
@@ -58,6 +62,7 @@ impl MarketplaceManager {
|
||||
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()),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -71,6 +76,16 @@ impl MarketplaceManager {
|
||||
&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()
|
||||
}
|
||||
@@ -329,6 +344,93 @@ pub fn compute_updates(
|
||||
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,
|
||||
@@ -560,4 +662,79 @@ mod tests {
|
||||
"slot is free after cancel"
|
||||
);
|
||||
}
|
||||
|
||||
#[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));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,9 +1,256 @@
|
||||
//! Pushes a project's marketplace payload into its container and runs the
|
||||
//! sync script there (spec §4).
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use super::payload::Payload;
|
||||
use crate::docker::exec::{exec_oneshot_as, exec_oneshot_streams_as, upload_bytes_to_container};
|
||||
use crate::models::marketplace::{SkippedItem, SyncReport};
|
||||
|
||||
/// Where the payload and the script are uploaded. Owned by `claude`.
|
||||
pub const INCOMING_DIR: &str = "/home/claude/.claude/triple-c/marketplace/incoming";
|
||||
|
||||
/// The sync script. Shipped with the app and uploaded on every sync, so a new
|
||||
/// app version reaches existing containers without an image migration.
|
||||
pub const SYNC_SCRIPT: &str = include_str!("sync.sh");
|
||||
|
||||
/// True once the entrypoint has finished: its last step execs this exact
|
||||
/// command line. Before that it may still be merging `settings.json` or running
|
||||
/// `claude update`, both of which the sync would race.
|
||||
const READY_PROBE: &str = "pgrep -x -f 'su -s /bin/bash claude -c exec sleep infinity' >/dev/null";
|
||||
const READY_TIMEOUT: Duration = Duration::from_secs(180);
|
||||
const READY_POLL: Duration = Duration::from_secs(2);
|
||||
|
||||
/// Run as root: `~/.claude` is a volume and `triple-c/` may not exist yet, and
|
||||
/// the uploads below are root-owned files in a directory `claude` must own so
|
||||
/// the script can delete them.
|
||||
const PREPARE_SCRIPT: &str = r#"set -e
|
||||
d=/home/claude/.claude/triple-c/marketplace/incoming
|
||||
mkdir -p "$d"
|
||||
chown -R claude:claude /home/claude/.claude/triple-c
|
||||
rm -f "$d/payload.tar" "$d/sync.sh""#;
|
||||
|
||||
fn sh(script: &str) -> Vec<String> {
|
||||
vec!["sh".to_string(), "-c".to_string(), script.to_string()]
|
||||
}
|
||||
|
||||
/// The readiness probe, run as root.
|
||||
fn ready_probe_cmd() -> Vec<String> {
|
||||
sh(READY_PROBE)
|
||||
}
|
||||
|
||||
/// The sync script invocation, run as `claude`.
|
||||
fn run_script_cmd() -> Vec<String> {
|
||||
vec!["sh".to_string(), format!("{INCOMING_DIR}/sync.sh")]
|
||||
}
|
||||
|
||||
fn run_script_env() -> Vec<String> {
|
||||
vec!["HOME=/home/claude".to_string()]
|
||||
}
|
||||
|
||||
async fn wait_until_ready(container_id: &str) -> Result<(), String> {
|
||||
let deadline = tokio::time::Instant::now() + READY_TIMEOUT;
|
||||
loop {
|
||||
let (_, code) = exec_oneshot_as(container_id, "root", ready_probe_cmd(), vec![]).await?;
|
||||
if code == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
if tokio::time::Instant::now() >= deadline {
|
||||
return Err(format!(
|
||||
"The container did not finish starting within {} seconds, so marketplace items \
|
||||
were not applied. They are applied on the next start, or with Apply now.",
|
||||
READY_TIMEOUT.as_secs()
|
||||
));
|
||||
}
|
||||
tokio::time::sleep(READY_POLL).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// The last `max` bytes of `text`, trimmed, never splitting a character.
|
||||
fn tail(text: &str, max: usize) -> &str {
|
||||
let text = text.trim();
|
||||
if text.len() <= max {
|
||||
return text;
|
||||
}
|
||||
let mut start = text.len() - max;
|
||||
while !text.is_char_boundary(start) {
|
||||
start += 1;
|
||||
}
|
||||
&text[start..]
|
||||
}
|
||||
|
||||
/// Wait for readiness, upload the payload and the script, run the script as
|
||||
/// `claude`, and return its report.
|
||||
pub async fn sync_container(container_id: &str, payload: &Payload) -> Result<SyncReport, String> {
|
||||
wait_until_ready(container_id).await?;
|
||||
|
||||
let (out, code) = exec_oneshot_as(container_id, "root", sh(PREPARE_SCRIPT), vec![]).await?;
|
||||
if code != 0 {
|
||||
return Err(format!(
|
||||
"Could not prepare the container for the marketplace sync: {}",
|
||||
tail(&out, 500)
|
||||
));
|
||||
}
|
||||
upload_bytes_to_container(
|
||||
container_id,
|
||||
INCOMING_DIR,
|
||||
"payload.tar",
|
||||
&payload.tar,
|
||||
0o644,
|
||||
)
|
||||
.await?;
|
||||
upload_bytes_to_container(
|
||||
container_id,
|
||||
INCOMING_DIR,
|
||||
"sync.sh",
|
||||
SYNC_SCRIPT.as_bytes(),
|
||||
0o755,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let (stdout, stderr, code) =
|
||||
exec_oneshot_streams_as(container_id, "claude", run_script_cmd(), run_script_env()).await?;
|
||||
parse_report(&stdout).map_err(|e| {
|
||||
format!(
|
||||
"The marketplace sync script failed (exit {code}): {e}. {}",
|
||||
tail(&stderr, 500)
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
/// The script's report is the last non-empty line of stdout.
|
||||
pub fn parse_report(stdout: &str) -> Result<SyncReport, String> {
|
||||
let line = stdout
|
||||
.lines()
|
||||
.rev()
|
||||
.map(str::trim)
|
||||
.find(|l| !l.is_empty())
|
||||
.ok_or_else(|| "the sync script printed no report".to_string())?;
|
||||
serde_json::from_str(line)
|
||||
.map_err(|e| format!("the sync script's report could not be read: {e}"))
|
||||
}
|
||||
|
||||
/// A sync never fails its caller: an error becomes a report that says so.
|
||||
pub fn report_from_result(r: Result<SyncReport, String>) -> SyncReport {
|
||||
let mut report = match r {
|
||||
Ok(report) => report,
|
||||
Err(e) => SyncReport {
|
||||
errors: vec![e],
|
||||
..Default::default()
|
||||
},
|
||||
};
|
||||
report.finished_at = chrono::Utc::now().to_rfc3339();
|
||||
report
|
||||
}
|
||||
|
||||
/// Items the host left out of the payload (invalid, missing from the cache, …)
|
||||
/// never reach the script, so the stored report lists them ahead of its own.
|
||||
pub fn with_payload_skips(mut report: SyncReport, payload_skipped: &[SkippedItem]) -> SyncReport {
|
||||
let mut skipped = payload_skipped.to_vec();
|
||||
skipped.append(&mut report.skipped);
|
||||
report.skipped = skipped;
|
||||
report
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn the_report_is_the_last_non_empty_stdout_line() {
|
||||
let out = "noise\n{\"installed\":[\"agent:a\"],\"errors\":[]}\n\n";
|
||||
let r = parse_report(out).unwrap();
|
||||
assert_eq!(r.installed, vec!["agent:a"]);
|
||||
assert!(r.skipped.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn missing_or_garbled_reports_are_errors() {
|
||||
assert!(parse_report("").unwrap_err().contains("no report"));
|
||||
assert!(parse_report("not json\n")
|
||||
.unwrap_err()
|
||||
.contains("could not be read"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_failed_sync_becomes_a_report() {
|
||||
// A failed sync becomes a report with the error in it — never an Err
|
||||
// that could propagate into container start.
|
||||
let r = report_from_result(Err("container went away".into()));
|
||||
assert_eq!(r.errors, vec!["container went away"]);
|
||||
assert!(!r.finished_at.is_empty());
|
||||
|
||||
let ok = report_from_result(Ok(SyncReport {
|
||||
installed: vec!["hook:h".into()],
|
||||
..Default::default()
|
||||
}));
|
||||
assert_eq!(ok.installed, vec!["hook:h"]);
|
||||
assert!(chrono::DateTime::parse_from_rfc3339(&ok.finished_at).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_embedded_script_is_the_sync_script() {
|
||||
assert!(SYNC_SCRIPT.starts_with("#!/bin/sh"));
|
||||
assert!(SYNC_SCRIPT.contains("MARKETPLACE_INCOMING"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn readiness_probes_the_entrypoints_final_exec() {
|
||||
assert_eq!(
|
||||
ready_probe_cmd(),
|
||||
vec![
|
||||
"sh",
|
||||
"-c",
|
||||
"pgrep -x -f 'su -s /bin/bash claude -c exec sleep infinity' >/dev/null"
|
||||
]
|
||||
);
|
||||
assert_eq!(READY_POLL, Duration::from_secs(2));
|
||||
assert_eq!(READY_TIMEOUT, Duration::from_secs(180));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_incoming_dir_is_prepared_for_claude() {
|
||||
// The uploads are root-owned, so the directory must exist and belong
|
||||
// to claude before they land (claude extracts and deletes them).
|
||||
assert!(PREPARE_SCRIPT.contains(INCOMING_DIR));
|
||||
assert!(PREPARE_SCRIPT.contains("mkdir -p"));
|
||||
assert!(PREPARE_SCRIPT.contains("chown -R claude:claude /home/claude/.claude/triple-c"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_script_runs_as_claude_with_home_set() {
|
||||
assert_eq!(
|
||||
run_script_cmd(),
|
||||
vec!["sh".to_string(), format!("{INCOMING_DIR}/sync.sh")]
|
||||
);
|
||||
assert_eq!(run_script_env(), vec!["HOME=/home/claude"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn payload_skips_come_before_the_scripts_own() {
|
||||
use crate::models::marketplace::SkippedItem;
|
||||
let payload_skip = SkippedItem {
|
||||
item: "agent:a".into(),
|
||||
reason: "invalid".into(),
|
||||
};
|
||||
let script_skip = SkippedItem {
|
||||
item: "hook:h".into(),
|
||||
reason: "no jq".into(),
|
||||
};
|
||||
let report = SyncReport {
|
||||
skipped: vec![script_skip.clone()],
|
||||
..Default::default()
|
||||
};
|
||||
let merged = with_payload_skips(report, &[payload_skip.clone()]);
|
||||
assert_eq!(merged.skipped, vec![payload_skip, script_skip]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn long_output_is_tailed_on_a_char_boundary() {
|
||||
assert_eq!(tail(" short \n", 10), "short");
|
||||
let s = format!("{}é", "x".repeat(20));
|
||||
let t = tail(&s, 1);
|
||||
assert!(s.ends_with(t));
|
||||
assert!(t.len() <= 2);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user