//! Native local snapshot management. No compilation, dlopen, provider or network.
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::{
collections::BTreeSet,
fs::{self, File, OpenOptions},
io::{Read, Write},
path::{Component, Path, PathBuf},
sync::Arc,
};
const RECEIPTS: &str = "resource-receipts.json";
const JOURNAL: &str = "resource-transaction.json";
const LEASE: &str = ".resource-lease";
const MAX_META: usize = 256 * 1024;
const MAX_FILES: usize = 2048;
const MAX_FILE: u64 = 64 * 1024 * 1024;
const MAX_BYTES: u64 = 256 * 1024 * 1024;
/// Static protocol description only: never resolves a root or opens metadata.
pub(crate) fn capabilities() -> Value {
json!({"version":1,"command":"resource","kinds":["package","skill","extension"],
"scopes":["project","global"],"explicitScopeRequired":true,
"operations":["list","inspect","install","update","reinstall","remove","recover"],
"localSnapshotsOnly":true,"build":false,"network":false,"hotApply":false,
"bounds":{"maxMetadataBytes":MAX_META,"maxResultBytes":MAX_META,"maxResourcesPerScope":128,
"maxInventoryEntries":MAX_FILES,"maxDepth":32,"maxFileBytes":MAX_FILE,"maxTotalBytes":MAX_BYTES}})
}
type Result<T> = std::result::Result<T, String>;
fn err(e: impl std::fmt::Display) -> String {
e.to_string()
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Scope {
#[default]
Project,
Global,
}
impl Scope {
pub fn is_project(&self) -> bool {
*self == Self::Project
}
pub fn root(self, cwd: &Path) -> Result<PathBuf> {
match self {
Self::Project => Ok(cwd.join(".rpi")),
Self::Global => crate::config::agent_dir().map_err(err),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
enum Kind {
Package,
Skill,
Extension,
}
impl Kind {
fn parse(s: &str) -> Result<Self> {
match s {
"package" => Ok(Self::Package),
"skill" => Ok(Self::Skill),
"extension" => Ok(Self::Extension),
_ => Err("kind must be package, skill or extension".into()),
}
}
fn dir(self) -> &'static str {
match self {
Self::Package => "packages",
Self::Skill => "skills",
Self::Extension => "extensions",
}
}
fn target(self, name: &str) -> PathBuf {
Path::new(self.dir()).join(if self == Self::Extension {
format!("{name}.{}", std::env::consts::DLL_EXTENSION)
} else {
name.into()
})
}
}
fn identifier(s: &str) -> Result<()> {
if s.is_empty()
|| s.len() > 96
|| !s.as_bytes()[0].is_ascii_alphanumeric()
|| !s
.bytes()
.all(|c| c.is_ascii_alphanumeric() || matches!(c, b'-' | b'_' | b'.'))
|| s.ends_with('.')
{
return Err("invalid resource identity (portable ASCII name required)".into());
}
reserved_identity(s)
}
// Existing native link records allow leading '_' and '.', unlike new managed
// identities. Accept that grammar only for safe on-disk package components.
fn legacy_package_identity(s: &str) -> Result<()> {
if !crate::packages::valid_name(s) {
return Err("invalid native package identity".into());
}
relative(Path::new(s))?;
reserved_identity(s)
}
fn reserved_identity(s: &str) -> Result<()> {
let stem = s.split('.').next().unwrap().to_ascii_uppercase();
if matches!(stem.as_str(), "CON" | "PRN" | "AUX" | "NUL")
|| (stem.len() == 4
&& (stem.starts_with("COM") || stem.starts_with("LPT"))
&& stem.as_bytes()[3].is_ascii_digit())
{
return Err("reserved resource identity".into());
}
Ok(())
}
fn relative(path: &Path) -> Result<()> {
if path.as_os_str().is_empty()
|| path
.components()
.any(|c| !matches!(c, Component::Normal(_)))
{
return Err("unsafe relative resource path".into());
}
for part in path {
let s = part.to_str().ok_or("non-UTF-8 resource path")?;
if s.len() > 128 || s.contains([':', '\\']) || s.ends_with(['.', ' ']) {
return Err("unsafe resource filename".into());
}
}
Ok(())
}
fn is_link(meta: &fs::Metadata) -> bool {
#[cfg(windows)]
{
use std::os::windows::fs::MetadataExt;
meta.file_type().is_symlink() || meta.file_attributes() & 0x400 != 0
}
#[cfg(not(windows))]
{
meta.file_type().is_symlink()
}
}
/// Validate every existing ancestor too, including Windows junctions.
pub(crate) fn checked_path(path: &Path) -> Result<()> {
if !path.is_absolute() {
return Err("resource path must be absolute".into());
}
let mut current = PathBuf::new();
for c in path.components() {
if matches!(c, Component::ParentDir) {
return Err("resource path traversal".into());
}
current.push(c);
// Prefix alone is not an addressable ancestor (notably `\\\\?\\C:`).
// Inspect the filesystem root once RootDir has made it absolute.
if matches!(c, Component::Prefix(_)) || !current.is_absolute() {
continue;
}
match fs::symlink_metadata(¤t) {
Ok(m) if is_link(&m) => {
return Err(format!(
"links/reparse points are not owned resource paths: {}",
current.display()
))
}
Ok(_) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => {
return Err(format!(
"checking resource ancestor {}: {e}",
current.display()
))
}
}
}
Ok(())
}
fn read_bytes(path: &Path, limit: usize) -> Result<Option<Vec<u8>>> {
checked_path(path)?;
let mut file = match File::open(path) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => return Err(err(e)),
};
if !file.metadata().map_err(err)?.is_file() {
return Err("resource metadata is not a regular file".into());
}
let mut bytes = Vec::new();
Read::by_ref(&mut file)
.take(limit as u64 + 1)
.read_to_end(&mut bytes)
.map_err(err)?;
if bytes.len() > limit {
return Err(format!("resource metadata exceeds {limit} bytes"));
}
Ok(Some(bytes))
}
fn read_json<T: for<'a> Deserialize<'a> + Default>(path: &Path) -> Result<T> {
match read_bytes(path, MAX_META)? {
None => Ok(T::default()),
Some(b) => {
serde_json::from_slice(&b).map_err(|e| format!("invalid {}: {e}", path.display()))
}
}
}
fn json_bytes(value: &impl Serialize) -> Result<Vec<u8>> {
let b = serde_json::to_vec_pretty(value).map_err(err)?;
if b.len() > MAX_META {
Err("resource metadata/output exceeds 256 KiB".into())
} else {
Ok(b)
}
}
/// Stable lock inode: never removed, no PID heuristics, no blocking wait.
#[derive(Debug)]
pub struct Lease {
_file: File,
}
#[derive(Debug)]
enum LeaseError {
Contended,
Other(String),
}
impl From<String> for LeaseError {
fn from(error: String) -> Self {
Self::Other(error)
}
}
impl From<LeaseError> for String {
fn from(error: LeaseError) -> Self {
match error {
LeaseError::Contended => {
"resource scope is in use (host or mutation lease); retry after owners exit".into()
}
LeaseError::Other(e) => format!("resource lease error: {e}"),
}
}
}
fn lock_error(error: std::io::Error) -> LeaseError {
let busy = fs2::lock_contended_error();
if error.raw_os_error().is_some() && error.raw_os_error() == busy.raw_os_error() {
LeaseError::Contended
} else {
LeaseError::Other(error.to_string())
}
}
impl Lease {
fn acquire(
root: &Path,
exclusive: bool,
create: bool,
) -> std::result::Result<Option<Self>, LeaseError> {
checked_path(root)?;
if create {
fs::create_dir_all(root).map_err(err)?;
}
let path = root.join(LEASE);
checked_path(&path)?;
let file = match OpenOptions::new()
.read(true)
.write(true)
.create(create)
.open(path)
{
Ok(f) => f,
Err(e) if !create && e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => return Err(LeaseError::Other(err(e))),
};
let result = if exclusive {
fs2::FileExt::try_lock_exclusive(&file)
} else {
fs2::FileExt::try_lock_shared(&file)
};
result.map_err(lock_error)?;
Ok(Some(Self { _file: file }))
}
}
impl Drop for Lease {
fn drop(&mut self) {
let _ = fs2::FileExt::unlock(&self._file);
}
}
pub(crate) fn mutation_lease(root: &Path) -> Result<Lease> {
let lease = Lease::acquire(root, true, true)?.ok_or("missing resource lease")?;
ensure_complete(root)?;
Ok(lease)
}
fn ensure_complete(root: &Path) -> Result<()> {
if read_bytes(&root.join(JOURNAL), MAX_META)?.is_some() {
return Err(
"incomplete resource transaction; run rpi resource recover with this scope".into(),
);
}
Ok(())
}
pub(crate) fn host_leases(cwd: &Path) -> Result<Arc<Vec<Lease>>> {
let mut roots = vec![Scope::Project.root(cwd)?, Scope::Global.root(cwd)?];
roots.sort();
roots.dedup();
let mut leases = Vec::new();
for root in roots {
leases.push(Lease::acquire(&root, false, true)?.ok_or("missing resource lease")?);
ensure_complete(&root)?;
}
Ok(Arc::new(leases))
}
fn observation_lease(root: &Path) -> Result<(Option<Lease>, bool)> {
// Retain the idle exclusive probe, or a shared lease alongside hosts.
// A mutator's exclusive lease refuses both; no unguarded half-state read.
match Lease::acquire(root, true, false) {
Ok(lease) => Ok((lease, false)),
Err(LeaseError::Contended) => Ok((Lease::acquire(root, false, false)?, true)),
Err(e) => Err(e.into()),
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
struct ResourceFile {
path: PathBuf,
bytes: u64,
sha256: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
struct Inventory {
files: Vec<ResourceFile>,
directories: Vec<PathBuf>,
sha256: String,
}
#[derive(Default)]
struct InventoryBudget {
entries: usize,
bytes: u64,
}
fn inventory(path: &Path) -> Result<Inventory> {
inventory_bounded(path, &mut InventoryBudget::default())
}
fn inventory_bounded(path: &Path, budget: &mut InventoryBudget) -> Result<Inventory> {
checked_path(path)?;
let mut files = Vec::new();
let mut directories = Vec::new();
fn walk(
root: &Path,
path: &Path,
files: &mut Vec<ResourceFile>,
dirs: &mut Vec<PathBuf>,
budget: &mut InventoryBudget,
depth: usize,
) -> Result<()> {
if depth > 32 {
return Err("resource recursion depth exceeds 32".into());
}
if budget.entries >= MAX_FILES {
return Err("resource inventory exceeds 2048 entries".into());
}
budget.entries += 1;
let meta = fs::symlink_metadata(path).map_err(err)?;
if is_link(&meta) {
return Err(format!(
"resource link/reparse point rejected: {}",
path.display()
));
}
let rel = path.strip_prefix(root).map_err(err)?;
if !rel.as_os_str().is_empty() {
relative(rel)?;
}
if meta.is_dir() {
if !rel.as_os_str().is_empty() {
dirs.push(rel.to_path_buf());
}
let remaining = MAX_FILES - budget.entries;
let mut entries = Vec::new();
for entry in fs::read_dir(path).map_err(err)?.take(remaining + 1) {
if entries.len() == remaining {
return Err("resource directory exceeds remaining entry budget".into());
}
entries.push(entry.map_err(err)?);
}
entries.sort_by_key(|e| e.file_name());
for e in entries {
walk(root, &e.path(), files, dirs, budget, depth + 1)?;
}
} else if meta.is_file() {
if meta.len() > MAX_FILE || budget.bytes + meta.len() > MAX_BYTES {
return Err("resource files exceed byte limit (64 MiB/file, 256 MiB total)".into());
}
let mut input = File::open(path).map_err(err)?;
let mut hash = Sha256::new();
let mut size = 0;
let mut buf = [0; 16384];
loop {
let n = input.read(&mut buf).map_err(err)?;
if n == 0 {
break;
}
size += n as u64;
budget.bytes += n as u64;
if size > MAX_FILE || budget.bytes > MAX_BYTES {
return Err("resource changed/exceeded byte limit".into());
}
hash.update(&buf[..n]);
}
if size != meta.len() {
return Err("resource changed while reading".into());
}
files.push(ResourceFile {
path: if rel.as_os_str().is_empty() {
PathBuf::from(".")
} else {
rel.into()
},
bytes: size,
sha256: format!("{:x}", hash.finalize()),
});
} else {
return Err("special resource file rejected".into());
}
Ok(())
}
walk(path, path, &mut files, &mut directories, budget, 0)?;
let sha256 = format!(
"{:x}",
Sha256::digest(serde_json::to_vec(&(&files, &directories)).map_err(err)?)
);
Ok(Inventory {
files,
directories,
sha256,
})
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Receipt {
kind: Kind,
name: String,
source: PathBuf,
version: Option<String>,
target: PathBuf,
inventory: Inventory,
}
#[derive(Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Receipts {
#[serde(default)]
resources: Vec<Receipt>,
}
fn receipts(root: &Path) -> Result<Receipts> {
let r: Receipts = read_json(&root.join(RECEIPTS))?;
let mut keys = BTreeSet::new();
if r.resources.len() > 128 {
return Err("resource receipts exceed 128 resources per scope".into());
}
for item in &r.resources {
identifier(&item.name)?;
if item.version.as_ref().is_some_and(|v| {
v.trim().is_empty() || v.len() > 128 || v.chars().any(char::is_control)
}) {
return Err("invalid resource receipt version".into());
}
if item.target != item.kind.target(&item.name)
|| !item.source.is_absolute()
|| !keys.insert(item.target.to_string_lossy().to_ascii_lowercase())
{
return Err("unsafe/duplicate resource receipt".into());
}
checked_path(&root.join(&item.target))?;
if item.inventory.files.len() + item.inventory.directories.len() > MAX_FILES {
return Err("oversized receipt inventory".into());
}
for file in &item.inventory.files {
if file.path != Path::new(".") {
relative(&file.path)?;
}
}
for dir in &item.inventory.directories {
relative(dir)?;
}
}
Ok(r)
}
#[derive(Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Packages {
#[serde(default)]
packages: Vec<crate::packages::ProjectPackage>,
}
fn packages(root: &Path) -> Result<Packages> {
let p: Packages = read_json(&root.join("packages.json"))?;
let mut names = BTreeSet::new();
for item in &p.packages {
legacy_package_identity(&item.name)?;
if !names.insert(item.name.to_ascii_lowercase()) {
return Err("duplicate native package identity".into());
}
let mut artifacts = BTreeSet::new();
for a in &item.artifacts {
relative(Path::new(a))?;
reserved_identity(a)?;
if Path::new(a).components().count() != 1 || !artifacts.insert(a.to_ascii_lowercase()) {
return Err("unsafe/duplicate native artifact".into());
}
}
}
Ok(p)
}
pub(crate) fn legacy_unowned(root: &Path, kind: &str, name: &str) -> Result<()> {
let kind = Kind::parse(kind)?;
if receipts(root)?
.resources
.iter()
.any(|r| r.kind == kind && r.name.eq_ignore_ascii_case(name))
{
return Err("resource is managed by rpi resource; use its update/remove commands".into());
}
if kind == Kind::Package {
legacy_package_identity(name)?;
let package = root.join(kind.target(name));
legacy_unowned_targets(root, &[package.clone(), package.join("extensions")])?;
}
Ok(())
}
fn path_identity(path: &Path) -> Result<PathBuf> {
let path = if path.is_absolute() {
path.to_path_buf()
} else {
std::env::current_dir().map_err(err)?.join(path)
};
checked_path(&path)?;
// Existing components are canonicalized; absent leaves still compare by
// their absolute addressed path. Reparse paths are rejected, not followed.
let mut ancestor = path.as_path();
let mut tail = Vec::new();
while !ancestor.try_exists().map_err(err)? {
tail.push(
ancestor
.file_name()
.ok_or("no path ancestor")?
.to_os_string(),
);
ancestor = ancestor.parent().ok_or("no path ancestor")?;
}
let mut identity = fs::canonicalize(ancestor).map_err(err)?;
for component in tail.into_iter().rev() {
identity.push(component);
}
if cfg!(windows) {
Ok(PathBuf::from(identity.to_string_lossy().to_lowercase()))
} else {
Ok(identity)
}
}
pub(crate) fn resource_owned_names(root: &Path, kind: &str) -> Result<BTreeSet<String>> {
let kind = Kind::parse(kind)?;
Ok(receipts(root)?
.resources
.into_iter()
.filter(|r| r.kind == kind)
.map(|r| r.name)
.collect())
}
pub(crate) fn legacy_unowned_targets(root: &Path, targets: &[PathBuf]) -> Result<()> {
let targets = targets
.iter()
.map(|t| path_identity(t))
.collect::<Result<Vec<_>>>()?;
for r in receipts(root)?.resources {
let owned = path_identity(&root.join(&r.target))?;
if targets.iter().any(|t| t == &owned || t.starts_with(&owned)) {
return Err("target owned by rpi resource; refusing legacy replacement/removal".into());
}
}
Ok(())
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Change {
target: PathBuf,
old: Option<Inventory>,
new: Option<Inventory>,
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Journal {
directory: String,
changes: Vec<Change>,
}
fn transaction_target(p: &Path) -> Result<()> {
relative(p)?;
if p == Path::new(RECEIPTS) || p == Path::new("packages.json") {
return Ok(());
}
let parts = p
.iter()
.map(|p| p.to_str().ok_or("non-UTF8 target"))
.collect::<std::result::Result<Vec<_>, _>>()?;
if parts.len() != 2 || !matches!(parts[0], "packages" | "skills" | "extensions") {
return Err("unsafe transaction target".into());
}
identifier(parts[1])
}
fn remove_path(path: &Path) -> Result<()> {
checked_path(path)?;
let m = fs::symlink_metadata(path).map_err(err)?;
if m.is_dir() {
fs::remove_dir_all(path).map_err(err)
} else if m.is_file() {
fs::remove_file(path).map_err(err)
} else {
Err("unsafe removal".into())
}
}
fn rollback(root: &Path, j: &Journal) -> Result<()> {
let dir = root.join(&j.directory);
checked_path(&dir)?;
// Validate the complete recovery plan before changing any file.
for (i, c) in j.changes.iter().enumerate() {
transaction_target(&c.target)?;
let target = root.join(&c.target);
let backup = dir.join(format!("old-{i}"));
if backup.try_exists().map_err(err)? {
if Some(inventory(&backup)?) != c.old {
return Err("transaction backup changed; manual recovery required".into());
}
if target.try_exists().map_err(err)? && Some(inventory(&target)?) != c.new {
return Err("transaction target changed; refusing recovery deletion".into());
}
} else if target.try_exists().map_err(err)? {
let observed = inventory(&target)?;
if c.old.as_ref() != Some(&observed) && c.new.as_ref() != Some(&observed) {
return Err("transaction target differs from old/new inventory".into());
}
if c.old.is_some() && c.old.as_ref() != Some(&observed) {
return Err("missing rollback backup; manual recovery required".into());
}
} else if c.old.is_some() {
return Err("missing original/backup; manual recovery required".into());
}
}
for (i, c) in j.changes.iter().enumerate().rev() {
let target = root.join(&c.target);
let backup = dir.join(format!("old-{i}"));
if backup.try_exists().map_err(err)? {
if target.try_exists().map_err(err)? {
remove_path(&target)?;
}
fs::rename(&backup, &target).map_err(err)?;
} else if c.old.is_none() && target.try_exists().map_err(err)? {
remove_path(&target)?;
}
}
// Journal last: a failed rollback stays diagnosable and retryable.
fs::remove_file(root.join(JOURNAL)).map_err(err)?;
fs::remove_dir_all(dir).map_err(err)
}
fn recover(root: &Path) -> Result<bool> {
let Some(bytes) = read_bytes(&root.join(JOURNAL), MAX_META)? else {
return Ok(false);
};
let j: Journal = serde_json::from_slice(&bytes).map_err(err)?;
let suffix = j
.directory
.strip_prefix(".resource-txn-")
.ok_or("invalid transaction directory")?;
uuid::Uuid::parse_str(suffix).map_err(err)?;
if j.changes.len() > 3 {
return Err("oversized transaction plan".into());
}
let mut seen = BTreeSet::new();
for c in &j.changes {
transaction_target(&c.target)?;
if !seen.insert(&c.target) {
return Err("duplicate transaction target".into());
}
}
rollback(root, &j)?;
Ok(true)
}
fn write_new(path: &Path, bytes: &[u8]) -> Result<()> {
let mut f = OpenOptions::new()
.write(true)
.create_new(true)
.open(path)
.map_err(err)?;
f.write_all(bytes).and_then(|_| f.sync_all()).map_err(err)
}
fn copy_snapshot(source: &Path, target: &Path, observed: &Inventory) -> Result<()> {
fn copy_file(source: &Path, target: &Path, expected: u64) -> Result<()> {
checked_path(source)?;
let input = File::open(source).map_err(err)?;
#[cfg(unix)]
let mode = {
use std::os::unix::fs::PermissionsExt;
// Only ordinary rwx permissions; never copy setuid/setgid/sticky.
input.metadata().map_err(err)?.permissions().mode() & 0o777
};
let mut output = OpenOptions::new()
.write(true)
.create_new(true)
.open(target)
.map_err(err)?;
let copied = std::io::copy(&mut input.take(expected + 1), &mut output).map_err(err)?;
if copied != expected {
return Err("resource changed during bounded copy".into());
}
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
output
.set_permissions(fs::Permissions::from_mode(mode))
.map_err(err)?;
}
output.sync_all().map_err(err)
}
if source.is_file() {
copy_file(
source,
target,
observed
.files
.first()
.ok_or("missing file inventory")?
.bytes,
)?;
} else {
fs::create_dir(target).map_err(err)?;
for d in &observed.directories {
fs::create_dir_all(target.join(d)).map_err(err)?;
}
for f in &observed.files {
let to = target.join(&f.path);
if let Some(p) = to.parent() {
fs::create_dir_all(p).map_err(err)?;
}
copy_file(&source.join(&f.path), &to, f.bytes)?;
}
}
if inventory(target)? != *observed || inventory(source)? != *observed {
return Err("resource source changed during snapshot copy".into());
}
Ok(())
}
/// Small same-filesystem transaction; injection is unit-test-only, never an env backdoor.
fn publish(
root: &Path,
changes: Vec<(PathBuf, Option<PathBuf>)>,
stage: tempfile::TempDir,
mut before: impl FnMut(usize) -> Result<()>,
) -> Result<Vec<String>> {
let directory = stage
.path()
.file_name()
.and_then(|s| s.to_str())
.ok_or("invalid staging directory")?
.to_owned();
let mut j = Journal {
directory,
changes: Vec::new(),
};
for (target, new) in &changes {
transaction_target(target)?;
let path = root.join(target);
checked_path(&path)?;
j.changes.push(Change {
target: target.clone(),
old: if path.try_exists().map_err(err)? {
Some(inventory(&path)?)
} else {
None
},
new: new.as_ref().map(|p| inventory(p)).transpose()?,
});
}
let journal_stage = stage.path().join("journal");
write_new(&journal_stage, &json_bytes(&j)?)?;
fs::rename(&journal_stage, root.join(JOURNAL)).map_err(err)?;
// Do not auto-delete backup evidence if interrupted or rollback itself fails.
let _owned = stage.keep();
let result = (|| {
for (i, (target, new)) in changes.iter().enumerate() {
before(i * 2)?;
let path = root.join(target);
if j.changes[i].old.is_some() {
fs::rename(&path, root.join(&j.directory).join(format!("old-{i}"))).map_err(err)?;
}
before(i * 2 + 1)?;
if let Some(new) = new {
fs::rename(new, &path).map_err(err)?;
}
}
before(changes.len() * 2)?;
Ok(())
})();
if let Err(error) = result {
return match rollback(root, &j) {
Ok(()) => Err(error),
Err(r) => Err(format!(
"{error}; rollback incomplete: {r}; run resource recover"
)),
};
}
// Commit point: removing the journal makes installed bytes authoritative.
// Orphan staging after this point contains only old bytes, never active data.
if let Err(error) = fs::remove_file(root.join(JOURNAL)) {
return match rollback(root,&j) {
Ok(())=>Err(format!("commit marker removal failed: {error}; rolled back")),
Err(rollback)=>Err(format!("commit marker removal failed: {error}; rollback incomplete: {rollback}; run resource recover")),
};
}
if let Err(e) = fs::remove_dir_all(root.join(&j.directory)) {
return Ok(vec![format!(
"committed; backup cleanup pending (OS error {:?}, {:?})",
e.raw_os_error(),
e.kind()
)]);
}
Ok(Vec::new())
}
fn stage_dir(root: &Path) -> Result<tempfile::TempDir> {
tempfile::Builder::new()
.prefix(&format!(".resource-txn-{}", uuid::Uuid::new_v4().simple()))
.rand_bytes(0)
.tempdir_in(root)
.map_err(err)
}
#[derive(Default)]
struct Options {
op: String,
kind: Option<Kind>,
name: Option<String>,
scope: Option<Scope>,
source: Option<PathBuf>,
version: Option<String>,
}
fn parse(args: &[String]) -> Result<Options> {
let mut o = Options {
op: args.first().ok_or("missing resource operation")?.clone(),
..Default::default()
};
let mut pos = Vec::new();
let mut i = 1;
while i < args.len() {
let a = &args[i];
if matches!(a.as_str(), "--scope" | "--path" | "--version") {
i += 1;
let v = args.get(i).ok_or("missing option value")?;
if v.starts_with('-') {
return Err("missing option value".into());
}
match a.as_str() {
"--scope" => {
if o.scope.is_some() {
return Err("duplicate scope".into());
}
o.scope = Some(match v.as_str() {
"project" => Scope::Project,
"global" => Scope::Global,
_ => return Err("scope must be project or global".into()),
});
}
"--path" => {
if o.source.replace(v.into()).is_some() {
return Err("duplicate path".into());
}
}
_ => {
if v.trim().is_empty()
|| v.len() > 128
|| v.chars().any(char::is_control)
|| o.version.replace(v.clone()).is_some()
{
return Err("invalid/duplicate version".into());
}
}
}
} else if a.starts_with('-') {
return Err(format!(
"unknown resource option {a}; builds require explicit native package link/install"
));
} else {
pos.push(a.clone());
}
i += 1;
}
if o.scope.is_none() {
return Err("explicit --scope project|global is required".into());
}
if matches!(o.op.as_str(), "list" | "recover") {
if !pos.is_empty() || o.source.is_some() || o.version.is_some() {
return Err("unexpected list/recover arguments".into());
}
} else if matches!(
o.op.as_str(),
"inspect" | "install" | "update" | "reinstall" | "remove"
) {
if pos.len() != 2 {
return Err("expected KIND NAME".into());
}
o.kind = Some(Kind::parse(&pos[0])?);
if o.op == "inspect" && o.kind == Some(Kind::Package) {
legacy_package_identity(&pos[1])?;
} else {
identifier(&pos[1])?;
}
o.name = Some(pos[1].clone());
let mutation = matches!(o.op.as_str(), "install" | "update" | "reinstall");
if mutation != o.source.is_some() || (!mutation && o.version.is_some()) {
return Err("install/update/reinstall require --path; inspect/remove reject it".into());
}
} else {
return Err("unknown resource operation".into());
}
Ok(o)
}
fn validate_source(kind: Kind, source: &Path, inv: &Inventory) -> Result<()> {
match kind {
Kind::Skill => {
if !source.is_dir() || !source.join("SKILL.md").is_file() {
return Err("standalone Skill requires a directory containing SKILL.md".into());
}
let text =
read_bytes(&source.join("SKILL.md"), 64 * 1024)?.ok_or("missing SKILL.md")?;
std::str::from_utf8(&text).map_err(err)?;
}
Kind::Extension => {
if !source.is_file()
|| source.extension().and_then(|s| s.to_str())
!= Some(std::env::consts::DLL_EXTENSION)
{
return Err("extension source must be a platform native library file".into());
}
}
Kind::Package => {
if !source.is_dir() {
return Err("package source must be a directory".into());
}
if !["extensions", "skills", "prompts", "themes"]
.iter()
.any(|s| source.join(s).is_dir())
&& inv.files.is_empty()
{
return Err("empty package".into());
}
for f in &inv.files {
if f.path.starts_with("extensions")
&& (f.path.components().count() != 2
|| f.path.extension().and_then(|s| s.to_str())
!= Some(std::env::consts::DLL_EXTENSION))
{
return Err(
"Package extensions must be direct platform native artifacts".into(),
);
}
}
}
}
Ok(())
}
fn explicit_source(path: &Path) -> Result<PathBuf> {
let addressed = std::env::current_dir().map_err(err)?.join(path);
let mut normalized = PathBuf::new();
for component in addressed.components() {
match component {
Component::CurDir => {}
Component::ParentDir => {
checked_path(&normalized)?;
if !normalized.pop() {
return Err("source escapes filesystem root".into());
}
}
other => {
normalized.push(other);
// A Windows drive prefix (`C:` / `\\\\?\\C:`) becomes
// absolute only after its RootDir component is appended.
if !matches!(other, Component::Prefix(_)) && normalized.is_absolute() {
checked_path(&normalized)?;
}
}
}
}
fs::canonicalize(normalized).map_err(err)
}
fn mutate(root: &Path, o: &Options) -> Result<Value> {
let _lease = mutation_lease(root)?;
let mut r = receipts(root)?;
let mut p = packages(root)?;
let native: Vec<crate::install::InstalledNativePackage> =
read_json(&root.join("native-packages.json"))?;
let kind = o.kind.ok_or("missing kind")?;
let name = o.name.as_deref().ok_or("missing name")?;
let target = kind.target(name);
let path = root.join(&target);
checked_path(&path)?;
let previous = r
.resources
.iter()
.find(|r| r.kind == kind && r.name.eq_ignore_ascii_case(name))
.cloned();
for owned in r.resources.iter().filter(|r| r.kind == Kind::Package) {
if p.packages.iter().find(|p| p.name == owned.name) != Some(&package_record(owned)) {
return Err(
"owned package receipt disagrees with native packages.json; refusing mutation"
.into(),
);
}
}
if kind == Kind::Extension
&& native.iter().any(|p| {
p.name.eq_ignore_ascii_case(name)
|| p.artifacts.iter().any(|a| {
target
.file_name()
.is_some_and(|f| f.to_string_lossy().eq_ignore_ascii_case(a))
})
})
{
return Err(
"extension identity/artifact owned by legacy native registry (even if missing)".into(),
);
}
if let Some(old) = &previous {
if old.name != name {
return Err("identity case collision".into());
}
if inventory(&path)? != old.inventory {
return Err("owned resource content changed; refusing overwrite/removal".into());
}
}
if o.op == "remove" && previous.is_none() {
if path.try_exists().map_err(err)? {
return Err("unmanaged destination: refusing removal".into());
}
return Ok(json!({"outcome":"absent","kind":kind,"name":name}));
}
if previous.is_none() && path.try_exists().map_err(err)? {
return Err("unmanaged destination conflict".into());
}
if kind == Kind::Package
&& previous.is_none()
&& p.packages.iter().any(|p| p.name.eq_ignore_ascii_case(name))
{
return Err("package identity owned by native package link".into());
}
if matches!(o.op.as_str(), "update" | "reinstall") && previous.is_none() {
return Err("resource is not owned/installed; use install".into());
}
let source = o
.source
.as_ref()
.map(|path| explicit_source(path))
.transpose()?;
if let Some(source) = &source {
let source_identity = path_identity(source)?;
let root_identity = path_identity(root)?;
if source_identity.starts_with(&root_identity)
|| root_identity.starts_with(&source_identity)
{
return Err("source/destination resource trees overlap".into());
}
}
let stage = stage_dir(root)?;
let mut next = None;
if let Some(source) = source {
// No Cargo metadata/build or native loader is called here.
let inv = inventory(&source)?;
validate_source(kind, &source, &inv)?;
let version = o
.version
.clone()
.or_else(|| previous.as_ref().and_then(|r| r.version.clone()));
let receipt = Receipt {
kind,
name: name.into(),
source,
version,
target: target.clone(),
inventory: inv,
};
if o.op == "install" {
if let Some(old) = &previous {
if old.source == receipt.source
&& old.version == receipt.version
&& old.inventory == receipt.inventory
{
return Ok(
json!({"outcome":"unchanged","resource":observe(root,o.scope.unwrap(),old,false)?}),
);
}
return Err("already installed; use update or reinstall with --path".into());
}
}
let staged = stage.path().join("snapshot");
copy_snapshot(&receipt.source, &staged, &receipt.inventory)?;
next = Some((receipt, staged));
}
r.resources.retain(|r| !(r.kind == kind && r.name == name));
if kind == Kind::Package {
p.packages.retain(|p| p.name != name);
if let Some((new, _)) = &next {
p.packages.push(package_record(new));
}
}
if let Some((receipt, _)) = &next {
r.resources.push(receipt.clone());
}
if r.resources.len() > 128 {
return Err("resource receipts exceed 128 resources per scope".into());
}
r.resources.sort_by_key(|r| r.target.clone());
p.packages.sort_by(|a, b| a.name.cmp(&b.name));
let receipt_stage = stage.path().join("receipts");
write_new(&receipt_stage, &json_bytes(&r)?)?;
let mut changes = vec![
(target, next.as_ref().map(|(_, p)| p.clone())),
(PathBuf::from(RECEIPTS), Some(receipt_stage)),
];
if kind == Kind::Package {
let staged = stage.path().join("packages");
write_new(&staged, &json_bytes(&p)?)?;
changes.push((PathBuf::from("packages.json"), Some(staged)));
}
let mut data = json!({"outcome":if o.op=="remove"{"removed"}else if previous.is_some(){"replaced"}else{"installed"},"kind":kind,"name":name,"diagnostics":[],"resource":next.as_ref().map(|(r,_)|observe_inventory(root,o.scope.unwrap(),r,false,Ok(r.inventory.clone()))).transpose()?});
// Bound the exact public envelope before publication. Reserve space for a
// bounded post-commit backup-cleanup diagnostic (not a false operation error).
if json_bytes(&success(o, data.clone()))
.map_err(|error| format!("resource success result rejected before publication: {error}"))?
.len()
> MAX_META - 4096
{
return Err("resource success result exceeds bounded publication budget".into());
}
let parent = path.parent().ok_or("no resource parent")?;
checked_path(parent)?;
fs::create_dir_all(parent).map_err(err)?;
let diagnostics = publish(root, changes, stage, |_| Ok(()))?;
data["diagnostics"] = json!(diagnostics);
Ok(data)
}
fn package_record(r: &Receipt) -> crate::packages::ProjectPackage {
crate::packages::ProjectPackage {
name: r.name.clone(),
version: r.version.clone().unwrap_or_default(),
source: r.source.to_string_lossy().into_owned(),
cargo_package: String::new(),
artifacts: r
.inventory
.files
.iter()
.filter(|f| f.path.starts_with("extensions"))
.filter_map(|f| {
f.path
.file_name()
.and_then(|s| s.to_str())
.map(str::to_owned)
})
.collect(),
}
}
fn observe(root: &Path, scope: Scope, r: &Receipt, busy: bool) -> Result<Value> {
observe_bounded(root, scope, r, busy, &mut InventoryBudget::default())
}
fn observe_bounded(
root: &Path,
scope: Scope,
r: &Receipt,
busy: bool,
budget: &mut InventoryBudget,
) -> Result<Value> {
let observed = inventory_bounded(&root.join(&r.target), budget);
observe_inventory(root, scope, r, busy, observed)
}
fn observe_inventory(
root: &Path,
scope: Scope,
r: &Receipt,
busy: bool,
observed: Result<Inventory>,
) -> Result<Value> {
let mut diagnostics =
vec!["native ABI compatibility is unverified; inspection never loads code".to_string()];
if observed.as_ref().ok() != Some(&r.inventory) {
diagnostics.push("installed content is missing or differs from ownership receipt".into());
}
if let Err(e) = &observed {
diagnostics.push(e.clone());
}
Ok(
json!({"kind":r.kind,"scope":scope,"identity":format!("{}:{}:{}",if scope==Scope::Project{"project"}else{"global"},r.kind.dir(),r.name),"name":r.name,"source":r.source,"version":r.version,"ownership":"resource","installed":true,"inUse":busy,"loaded":null,"applied":null,"path":root.join(&r.target),"selectable":selectable(root,scope,r),"recordedDigest":r.inventory.sha256,"observed":observed.ok(),"diagnostics":diagnostics}),
)
}
fn selectable(root: &Path, scope: Scope, r: &Receipt) -> Value {
match r.kind {
Kind::Package => json!({"main_package":{"name":r.name,"scope":scope,"instructions":[]}}),
Kind::Skill => json!({"skills":[root.join(&r.target)]}),
Kind::Extension => json!({"extensions":[root.join(&r.target)]}),
}
}
fn catalog(root: &Path, o: &Options) -> Result<Value> {
checked_path(root)?;
let (_observation_lease, busy) = observation_lease(root)?;
let r = receipts(root)?;
let p = packages(root)?;
let native: Vec<crate::install::InstalledNativePackage> =
read_json(&root.join("native-packages.json"))?;
let scope = o.scope.unwrap();
let mut items = Vec::new();
let mut owned = BTreeSet::new();
let mut budget = InventoryBudget::default();
for receipt in &r.resources {
owned.insert(receipt.target.clone());
items.push(observe_bounded(root, scope, receipt, busy, &mut budget)?);
}
for kind in [Kind::Package, Kind::Skill, Kind::Extension] {
let directory = root.join(kind.dir());
checked_path(&directory)?;
let entries = match fs::read_dir(directory) {
Ok(e) => e,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
Err(e) => return Err(err(e)),
};
for entry in entries {
let entry = entry.map_err(err)?;
let target = Path::new(kind.dir()).join(entry.file_name());
if owned.contains(&target) {
continue;
}
if items.len() >= MAX_FILES {
return Err("catalog exceeds entry limit".into());
}
let name = if kind == Kind::Extension {
entry
.path()
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("")
.to_string()
} else {
entry
.file_name()
.to_str()
.ok_or("non UTF8 candidate")?
.to_string()
};
let pkg = p
.packages
.iter()
.find(|p| kind == Kind::Package && p.name == name);
let ext = native.iter().find(|p| {
kind == Kind::Extension
&& p.artifacts
.iter()
.any(|a| entry.file_name() == std::ffi::OsStr::new(a))
});
let inv = inventory_bounded(&entry.path(), &mut budget);
let diagnostics = match &inv {
Ok(_) => vec!["unverified candidate; no code executed".to_string()],
Err(e) => vec![e.clone()],
};
items.push(json!({"kind":kind,"scope":scope,"identity":format!("{}:{}:{}",if scope==Scope::Project{"project"}else{"global"},kind.dir(),name),"name":name,"source":pkg.map(|p|p.source.clone()).or_else(||ext.and_then(|p|p.source.clone())),"version":pkg.and_then(|p|if p.version.is_empty(){None}else{Some(&p.version)}).or_else(||ext.filter(|p|!p.version.trim().is_empty()).map(|p|&p.version)),"ownership":if pkg.is_some()||ext.is_some(){"legacy-native"}else{"unmanaged"},"installed":pkg.is_some()||ext.is_some(),"inUse":busy,"loaded":null,"applied":null,"path":entry.path(),"observed":inv.ok(),"diagnostics":diagnostics}));
}
}
if o.op == "inspect" {
items.retain(|item| item["name"] == json!(o.name) && item["kind"] == json!(o.kind));
if items.is_empty() {
return Err("resource not found".into());
}
}
items.sort_by_key(|i| i["identity"].as_str().unwrap_or("").to_string());
// When no persistent lock existed initially, first publication could have
// started during traversal. Its lock inode is never unlinked: reject and
// retry rather than returning any potentially mixed observation.
if _observation_lease.is_none() && root.join(LEASE).try_exists().map_err(err)? {
return Err("resource ownership initialized during observation; retry catalog".into());
}
Ok(
json!({"outcome":"observed","resources":items,"diagnostics":if root.join(JOURNAL).try_exists().map_err(err)?{vec!["incomplete transaction; run resource recover"]}else{vec![]}}),
)
}
fn success(o: &Options, data: Value) -> Value {
json!({"version":1,"success":true,"operation":o.op,"scope":o.scope,"kind":o.kind,"name":o.name,"data":data})
}
pub fn run(args: &[String]) -> i32 {
if args.is_empty() || args.iter().any(|a| matches!(a.as_str(), "--help" | "-h")) {
println!("rpi resource list|inspect|install|update|reinstall|remove|recover --scope project|global\nMutations: KIND NAME --path PATH [--version TEXT]; remove/inspect: KIND NAME.\nLocal snapshots only; no builds or native registration. JSON v1, 256 KiB bound.");
return 0;
}
let result: Result<Value> = (|| {
let o = parse(args)?;
let cwd = std::env::current_dir().map_err(err)?;
let root = o.scope.unwrap().root(&cwd)?;
let data = if matches!(o.op.as_str(), "list" | "inspect") {
catalog(&root, &o)?
} else if o.op == "recover" {
let _lease = Lease::acquire(&root, true, true)?;
json!({"outcome":if recover(&root)?{"recovered"}else{"unchanged"}})
} else {
mutate(&root, &o)?
};
Ok(success(&o, data))
})();
match result.and_then(|v| json_bytes(&v)) {
Ok(bytes) => {
println!("{}", String::from_utf8_lossy(&bytes));
0
}
Err(e) => {
let bounded = e.chars().take(2048).collect::<String>();
let context = parse(args).ok();
println!(
"{}",
json!({"version":1,"success":false,"operation":context.as_ref().map(|o|&o.op),"scope":context.as_ref().and_then(|o|o.scope),"kind":context.as_ref().and_then(|o|o.kind),"name":context.as_ref().and_then(|o|o.name.as_ref()),"error":bounded})
);
1
}
}
}
#[cfg(test)]
mod regressions;
#[cfg(test)]
mod tests;