diff --git a/README.md b/README.md
index fabc674..72b5e4b 100644
--- a/README.md
+++ b/README.md
@@ -224,3 +224,5 @@ python scripts/neon_genie.py do run --recipe commercial --out out/neon-genie/aud
Neon Genie v3.26.0 · advice only · evidence before invention
+
+Source, profile, and migration rules: `references/source-and-upgrades.md`.
diff --git a/SKILL.md b/SKILL.md
index e273484..9c2c30f 100644
--- a/SKILL.md
+++ b/SKILL.md
@@ -630,3 +630,5 @@ Hermes Hub installs only `SKILL.md` plus **explicitly path-referenced** files un
Full tree also keeps root schemas, profiles, evals, VERSION, and manifest for clone/`./install.sh` installs (scripts resolve either layout via `scripts/paths.py`).
+
+Source, profile, and migration rules: `references/source-and-upgrades.md`.
diff --git a/install.sh b/install.sh
index a9099c8..46bf6b6 100755
--- a/install.sh
+++ b/install.sh
@@ -5,37 +5,30 @@
set -euo pipefail
-TARGET_BASE="${HOME}/.hermes/skills"
+TARGET_BASE="${HERMES_HOME:-$HOME/.hermes}/skills"
DEST="${TARGET_BASE}/neon-genie"
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+DRY_RUN=0
+while [[ $# -gt 0 ]]; do
+ case "$1" in
+ --dry-run) DRY_RUN=1; shift ;;
+ --target) [[ $# -ge 2 ]] || { printf 'Missing target\n' >&2; exit 2; }; DEST="$2"; shift 2 ;;
+ --target=*) DEST="${1#--target=}"; shift ;;
+ -h|--help) printf 'Usage: install.sh [--target DIR] [--dry-run]\n'; exit 0 ;;
+ *) printf 'Unknown option: %s\n' "$1" >&2; exit 2 ;;
+ esac
+done
+ python3 - "$DEST" <<'PY'
+import sys
+from pathlib import Path
+p = Path(sys.argv[1]).expanduser().absolute()
+if not sys.argv[1].strip() or any(q.is_symlink() for q in (p, *p.parents)):
+ raise SystemExit("refusing empty or symlinked target path")
+PY
echo "Installing Neon Genie to: ${DEST}"
-mkdir -p "${TARGET_BASE}"
-
-if [ -d "${DEST}" ]; then
- echo "Existing installation found. Backing up to ${DEST}.bak"
- rm -rf "${DEST}.bak"
- mv "${DEST}" "${DEST}.bak"
-fi
-
-mkdir -p "${DEST}"
-# Copy skill tree; exclude VCS metadata
-if command -v rsync >/dev/null 2>&1; then
- rsync -a --exclude '.git' --exclude '.gitignore' "${ROOT}/" "${DEST}/"
-else
- tar -C "${ROOT}" --exclude '.git' -cf - . | tar -C "${DEST}" -xf -
+if [[ $DRY_RUN -eq 1 ]]; then
+ echo "DRY RUN: no files changed"
+ exit 0
fi
-
-chmod +x "${DEST}/scripts/"*.py 2>/dev/null || true
-chmod +x "${DEST}/install.sh" 2>/dev/null || true
-
-echo ""
-echo "Neon Genie installed successfully."
-echo "Location: ${DEST}"
-echo ""
-echo "Next steps:"
-echo " 1. Restart Hermes or reload skills."
-echo " 2. Validate: python ${DEST}/scripts/validate_hermes_skill.py"
-echo " 3. Try triggers like: 'product audit', 'zero option', 'wayfinder handoff'"
-echo ""
-echo "The skill works standalone inside Hermes (advisory only)."
+exec python3 "${ROOT}/scripts/install_transaction.py" "$ROOT" "$DEST" neon-genie
diff --git a/references/source-and-upgrades.md b/references/source-and-upgrades.md
new file mode 100644
index 0000000..ec14b69
--- /dev/null
+++ b/references/source-and-upgrades.md
@@ -0,0 +1,58 @@
+# Source identity and upgrades
+
+This distribution is `scrimshawlife-ctrl/NeonGenie`, version `3.26.0`, based on commit
+`9e7e6d388a77a9f6103fba6f982457afc873084b` before the local audit-fix commit. Record the actual installed commit
+with `git rev-parse HEAD` before installation; do not use a version string alone
+as a source identity. Root and hub packaging are distribution surfaces, not a
+claim that organizational, personal, or legacy embedded variants are identical.
+No personal version is deprecated by this change.
+
+Before switching sources, stop runtime writers, record current source/commit,
+review the destination under `${HERMES_HOME:-$HOME/.hermes}`, compare contracts,
+and retain a separate backup of outputs, sessions, and local customization.
+Do not install two different contracts with the same skill name into one profile.
+A successful compatibility check does not grant deployment/publication authority.
+
+The bundled LICENSE controls this distribution. No license grant is changed by
+these fixes; earlier grants and third-party notices remain intact.
+
+The installer validates a fresh stage on the target filesystem before replacement,
+then reads back the activated contract/version/provenance receipt. Checks use an
+isolated temporary home. Legacy `out` contents (including an out symlink) survive.
+Backups are unique, keyed by canonical destination under the active Hermes home's
+`backups///`; `.install-provenance.json` records source, base
+commit, dirty status, version, destination, and skipped checks. Stop writers during
+upgrades. Two renames have an absent-target window: this is NOT crash-atomic.
+On a reported recovery failure, retain the printed recovery directory and backup;
+do not delete it or retry blindly. Inspect the exact target and restore from the
+named backup only after validating it. No automatic source migration is implied.
+
+## Interrupted installs and stale locks
+
+SIGKILL or power loss can leave `..install-lock` beside the target.
+Locks are never automatically reclaimed: age, an empty directory, or a reused PID
+cannot prove that no installer is active. To recover:
+
+1. Stop install launchers and runtime writers. Confirm no installer is active on
+ this target (including other sessions/hosts sharing the filesystem).
+2. Inspect the exact target, sibling `.-stage-*` recovery directories
+ (especially `previous` and `failed-package`), and target-keyed backups. Retain
+ all recovery data until the original installation and outputs are accounted
+ for. Restore/validate a complete package first if activation was interrupted.
+3. Only then set `lock` to the exact path printed in the error and run
+ `rmdir -- "$lock"`. This removes only an empty lock; never use recursive
+ deletion or remove a lock whose owner/activity is uncertain. Keep launchers
+ stopped through inspection and removal to avoid races, then retry installation.
+
+Backup copies are staged in `.backup-incomplete-*` outside the target's selectable
+backup directory and published by rename after copying and writing the destination
+record. Hard termination can leave these incomplete staging trees; never select
+one for rollback. Inspect them manually after quiescing writers. Rollback skips
+legacy partial entries without a valid matching destination record.
+
+Git is optional. Archive sources, failed Git lookups, and unrelated enclosing
+worktrees record unknown Git fields as JSON `null`, never a false clean claim.
+A source-root worktree or the tracked `skills/neon-genie` hub with matching root
+contract/version supplies Git provenance. Receipts distinguish
+`source_repository_root` from `source_subdirectory` (`.` or `skills/neon-genie`).
+A dirty status lookup failure remains `null` even in a recognized worktree.
diff --git a/scripts/install_transaction.py b/scripts/install_transaction.py
new file mode 100644
index 0000000..c11124e
--- /dev/null
+++ b/scripts/install_transaction.py
@@ -0,0 +1,295 @@
+"""Staged, checked upgrades with target-keyed backups and recovery.
+
+The two renames have an absent-target window; this is not crash-atomic. Stop
+runtime writers during upgrades. A retained recovery directory requires operator
+inspection. No live profile is used by validation subprocesses.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import uuid
+from pathlib import Path
+
+CHECKS = {
+ "neon-genie": [("validate_hermes_skill.py",), ("neon_genie.py", "do", "check")],
+ "sigil-forge": [("validate_hermes_skill.py",), ("sigil_forge.py", "check")],
+ "hyperlex": [("hyperlex.py", "check"), ("hyperlex.py", "smoke")],
+}
+IGNORE = shutil.ignore_patterns(
+ ".git",
+ "skills",
+ "out",
+ ".venv",
+ "__pycache__",
+ ".pytest_cache",
+ ".mypy_cache",
+ ".ruff_cache",
+ "*.pyc",
+ ".worktrees",
+ "graft",
+ ".env",
+ ".env.*",
+ ".hermes",
+ "*.pem",
+ "*.key",
+ ".superpowers",
+ "superpowers",
+)
+
+
+def install(source: Path, target: Path, kind: str, skip_checks: bool = False) -> None:
+ # Check the lexical final component before resolve follows a dangling link.
+ if any(p.is_symlink() for p in (target.absolute(), *target.absolute().parents)):
+ raise ValueError("refusing symlink target")
+ source, target = source.resolve(), target.resolve()
+ if source == target or source in target.parents or target in source.parents:
+ raise ValueError("source and target must not overlap")
+ home = Path(os.environ.get("HERMES_HOME") or str(Path.home() / ".hermes")).resolve()
+ if target in (Path("/"), Path.home().resolve(), home, home / "skills"):
+ raise ValueError("refusing installation root")
+ if target.exists() and not target.is_dir():
+ raise ValueError("target must be a directory")
+ key = hashlib.sha256(str(target).encode()).hexdigest()[:20]
+ backups = home / "backups" / kind / key
+ if target == backups or target in backups.parents or backups in target.parents:
+ raise ValueError("backup and target must not overlap")
+ # Reject payload links rather than shipping aliases into private source state.
+ for directory, dirs, files in os.walk(source, followlinks=False):
+ ignored = IGNORE(directory, dirs + files)
+ dirs[:] = [d for d in dirs if d not in ignored]
+ for name in dirs + [f for f in files if f not in ignored]:
+ if (Path(directory) / name).is_symlink():
+ raise ValueError(f"source payload symlink is not portable: {name}")
+ target.parent.mkdir(parents=True, exist_ok=True)
+ lock = target.parent / ("." + target.name + ".install-lock")
+ try:
+ lock.mkdir() # exclusive; never remove another installer's lock
+ except FileExistsError as exc:
+ raise FileExistsError(
+ f"Install lock exists: {lock}. Do not reclaim automatically. "
+ "Stop launchers and confirm no installer or runtime writer is active; "
+ "inspect the target, sibling stage/recovery directories and backups. "
+ "Only after resolving recovery, use rmdir on this exact empty lock "
+ "(never recursive removal), then retry. See references/source-and-upgrades.md."
+ ) from exc
+ workspace = None
+ retain_recovery = False
+ try:
+ workspace = Path(
+ tempfile.mkdtemp(prefix="." + kind + "-stage-", dir=target.parent)
+ )
+ stage = workspace / "package"
+ shutil.copytree(source, stage, ignore=IGNORE)
+ check_home = workspace / "check-home"
+ check_home.mkdir()
+ env = {
+ k: v
+ for k, v in os.environ.items()
+ if not k.startswith(("HYPERLEX_", "SIGIL_FORGE_"))
+ }
+ env.update(
+ HOME=str(check_home),
+ HERMES_HOME=str(check_home / ".hermes"),
+ HERMES_SKILL_DIR=str(stage),
+ SIGIL_FORGE_STATE_DIR=str(check_home / "state"),
+ PYTHONDONTWRITEBYTECODE="1",
+ )
+ if not skip_checks:
+ for command in CHECKS[kind]:
+ subprocess.run(
+ [sys.executable, str(stage / "scripts" / command[0]), *command[1:]],
+ cwd=check_home,
+ env=env,
+ check=True,
+ )
+ if (stage / "out").exists():
+ shutil.rmtree(stage / "out") # NEW staging data only
+ old_out = target / "out"
+ if old_out.is_symlink():
+ (stage / "out").symlink_to(os.readlink(old_out), target_is_directory=True)
+ elif old_out.exists():
+ shutil.copytree(old_out, stage / "out", symlinks=True)
+
+ def git(*args: str) -> str | None:
+ try:
+ r = subprocess.run(
+ check=False,
+ args=["git", "-C", str(source), *args],
+ capture_output=True,
+ text=True,
+ env={
+ k: v for k, v in os.environ.items() if not k.startswith("GIT_")
+ },
+ )
+ except OSError:
+ return None # Git is optional for archive/Python-only installs.
+ return r.stdout.strip() if r.returncode == 0 else None
+
+ # Git searches parents: accept the source root or the tracked Neon hub,
+ # not arbitrary archives nested in an unrelated worktree.
+ top = git("rev-parse", "--show-toplevel")
+ repo_root = Path(top).resolve() if top is not None else None
+ own_checkout = repo_root == source
+ subtree = "." if own_checkout else None
+ if (
+ repo_root is not None
+ and kind == "neon-genie"
+ and source == repo_root / "skills/neon-genie"
+ ):
+ tracked = git(
+ "ls-files",
+ "--error-unmatch",
+ "--",
+ "SKILL.md",
+ "VERSION",
+ str(repo_root / "SKILL.md"),
+ str(repo_root / "VERSION"),
+ )
+ try:
+ # Root/hub grants can legitimately differ; compare the contract
+ # apart from its license metadata, without changing either grant.
+ root_contract = (repo_root / "SKILL.md").read_text().splitlines()
+ hub_contract = (source / "SKILL.md").read_text().splitlines()
+ matching_root = [
+ line for line in root_contract if not line.startswith("license:")
+ ] == [
+ line for line in hub_contract if not line.startswith("license:")
+ ] and (repo_root / "VERSION").read_bytes() == (
+ source / "VERSION"
+ ).read_bytes()
+ except (OSError, UnicodeError):
+ matching_root = False
+ if tracked is not None and matching_root:
+ own_checkout = True
+ subtree = "skills/neon-genie"
+ dirty = git("status", "--porcelain") if own_checkout else None
+ receipt = {
+ "source": str(source),
+ "source_repository_root": str(repo_root) if own_checkout else None,
+ "source_subdirectory": subtree,
+ "repository": git("remote", "get-url", "origin") if own_checkout else None,
+ "source_commit": git("rev-parse", "HEAD") if own_checkout else None,
+ "source_dirty": bool(dirty) if dirty is not None else None,
+ "version": (source / "VERSION").read_text().strip(),
+ "destination": str(target),
+ "status": "UNVERIFIED" if skip_checks else "VALIDATED",
+ "checks_skipped": list(CHECKS[kind]) if skip_checks else [],
+ "validation": "staged runtime; activated contract/version read-back",
+ }
+ receipt_text = json.dumps(receipt, indent=2) + "\n"
+ (stage / ".install-provenance.json").write_text(receipt_text)
+ expected = {
+ p: (stage / p).read_bytes()
+ for p in ("SKILL.md", "VERSION", ".install-provenance.json")
+ }
+ backup = None
+ if target.exists():
+ backups.mkdir(parents=True, exist_ok=True)
+ backup = backups / uuid.uuid4().hex
+ # Incomplete copies live outside the selectable backup namespace.
+ # Publish only after the payload and destination record are complete.
+ with tempfile.TemporaryDirectory(
+ prefix=".backup-incomplete-", dir=backups.parent
+ ) as tmp:
+ pending = Path(tmp) / "backup-payload"
+ shutil.copytree(target, pending, symlinks=True)
+ marker = pending / ".backup-target.json"
+ marker.unlink(missing_ok=True)
+ marker.write_text(json.dumps({"destination": str(target)}) + "\n")
+ os.replace(pending, backup)
+ displaced = workspace / "previous"
+ # Detect a concurrent change of target identity since staging began.
+ if any(p.is_symlink() for p in (target.absolute(), *target.absolute().parents)):
+ raise ValueError("target became a symlink during staging")
+ try:
+ if target.exists():
+ os.replace(target, displaced)
+ os.replace(stage, target)
+ for relative, content in expected.items():
+ if (target / relative).read_bytes() != content:
+ raise OSError(f"activation read-back mismatch: {relative}")
+ except BaseException as original:
+ try:
+ # Infer completed renames from disk even if replace raised after
+ # its side effect. A still-staged package means target is not new.
+ if not stage.exists() and target.exists():
+ os.replace(target, workspace / "failed-package")
+ if displaced.exists():
+ os.replace(displaced, target)
+ except BaseException as recovery_error:
+ retain_recovery = True
+ raise RuntimeError(
+ f"recovery required at {workspace}; backup {backup}; "
+ f"activation: {original}; restoration: {recovery_error}"
+ ) from recovery_error
+ raise
+ print(f"Installed {kind}: {target} ({receipt['status']})")
+ if backup:
+ print(f"Backup: {backup}")
+ finally:
+ if workspace is not None and not retain_recovery:
+ shutil.rmtree(workspace)
+ lock.rmdir()
+
+
+def rollback(target: Path, kind: str, skip_checks: bool = False) -> None:
+ if any(p.is_symlink() for p in (target.absolute(), *target.absolute().parents)):
+ raise ValueError("refusing symlink target")
+ target = target.resolve()
+ home = Path(os.environ.get("HERMES_HOME") or str(Path.home() / ".hermes")).resolve()
+ key = hashlib.sha256(str(target).encode()).hexdigest()[:20]
+ backups = home / "backups" / kind / key
+ candidates = sorted(
+ backups.glob("*"), key=lambda p: p.lstat().st_mtime_ns, reverse=True
+ )
+ backup = None
+ for candidate in candidates:
+ marker = candidate / ".backup-target.json"
+ if candidate.is_symlink() or not candidate.is_dir() or marker.is_symlink():
+ continue
+ try:
+ record = json.loads(marker.read_text())
+ except (OSError, ValueError):
+ continue
+ if isinstance(record, dict) and record.get("destination") == str(target):
+ backup = candidate
+ break
+ if backup is None:
+ raise ValueError(
+ f"no target-bound backup for {target}; legacy backups require manual review"
+ )
+ # Reuse staged runtime checks, backup, activated read-back and failed-rename recovery.
+ install(backup, target, kind, skip_checks)
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("source", type=Path)
+ parser.add_argument("target")
+ parser.add_argument("kind", choices=CHECKS)
+ parser.add_argument("--skip-checks", action="store_true")
+ parser.add_argument("--rollback", action="store_true")
+ args = parser.parse_args()
+ try:
+ if not args.target.strip():
+ raise ValueError("empty target")
+ if args.rollback:
+ rollback(Path(args.target), args.kind, args.skip_checks)
+ else:
+ install(args.source, Path(args.target), args.kind, args.skip_checks)
+ except (OSError, ValueError, RuntimeError, subprocess.CalledProcessError) as exc:
+ print(f"Installation failed: {exc}", file=sys.stderr)
+ return 1
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/skills/neon-genie/SKILL.md b/skills/neon-genie/SKILL.md
index e273484..9c2c30f 100644
--- a/skills/neon-genie/SKILL.md
+++ b/skills/neon-genie/SKILL.md
@@ -630,3 +630,5 @@ Hermes Hub installs only `SKILL.md` plus **explicitly path-referenced** files un
Full tree also keeps root schemas, profiles, evals, VERSION, and manifest for clone/`./install.sh` installs (scripts resolve either layout via `scripts/paths.py`).
+
+Source, profile, and migration rules: `references/source-and-upgrades.md`.
diff --git a/skills/neon-genie/install.sh b/skills/neon-genie/install.sh
index a9099c8..46bf6b6 100755
--- a/skills/neon-genie/install.sh
+++ b/skills/neon-genie/install.sh
@@ -5,37 +5,30 @@
set -euo pipefail
-TARGET_BASE="${HOME}/.hermes/skills"
+TARGET_BASE="${HERMES_HOME:-$HOME/.hermes}/skills"
DEST="${TARGET_BASE}/neon-genie"
ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+DRY_RUN=0
+while [[ $# -gt 0 ]]; do
+ case "$1" in
+ --dry-run) DRY_RUN=1; shift ;;
+ --target) [[ $# -ge 2 ]] || { printf 'Missing target\n' >&2; exit 2; }; DEST="$2"; shift 2 ;;
+ --target=*) DEST="${1#--target=}"; shift ;;
+ -h|--help) printf 'Usage: install.sh [--target DIR] [--dry-run]\n'; exit 0 ;;
+ *) printf 'Unknown option: %s\n' "$1" >&2; exit 2 ;;
+ esac
+done
+ python3 - "$DEST" <<'PY'
+import sys
+from pathlib import Path
+p = Path(sys.argv[1]).expanduser().absolute()
+if not sys.argv[1].strip() or any(q.is_symlink() for q in (p, *p.parents)):
+ raise SystemExit("refusing empty or symlinked target path")
+PY
echo "Installing Neon Genie to: ${DEST}"
-mkdir -p "${TARGET_BASE}"
-
-if [ -d "${DEST}" ]; then
- echo "Existing installation found. Backing up to ${DEST}.bak"
- rm -rf "${DEST}.bak"
- mv "${DEST}" "${DEST}.bak"
-fi
-
-mkdir -p "${DEST}"
-# Copy skill tree; exclude VCS metadata
-if command -v rsync >/dev/null 2>&1; then
- rsync -a --exclude '.git' --exclude '.gitignore' "${ROOT}/" "${DEST}/"
-else
- tar -C "${ROOT}" --exclude '.git' -cf - . | tar -C "${DEST}" -xf -
+if [[ $DRY_RUN -eq 1 ]]; then
+ echo "DRY RUN: no files changed"
+ exit 0
fi
-
-chmod +x "${DEST}/scripts/"*.py 2>/dev/null || true
-chmod +x "${DEST}/install.sh" 2>/dev/null || true
-
-echo ""
-echo "Neon Genie installed successfully."
-echo "Location: ${DEST}"
-echo ""
-echo "Next steps:"
-echo " 1. Restart Hermes or reload skills."
-echo " 2. Validate: python ${DEST}/scripts/validate_hermes_skill.py"
-echo " 3. Try triggers like: 'product audit', 'zero option', 'wayfinder handoff'"
-echo ""
-echo "The skill works standalone inside Hermes (advisory only)."
+exec python3 "${ROOT}/scripts/install_transaction.py" "$ROOT" "$DEST" neon-genie
diff --git a/skills/neon-genie/references/source-and-upgrades.md b/skills/neon-genie/references/source-and-upgrades.md
new file mode 100644
index 0000000..ec14b69
--- /dev/null
+++ b/skills/neon-genie/references/source-and-upgrades.md
@@ -0,0 +1,58 @@
+# Source identity and upgrades
+
+This distribution is `scrimshawlife-ctrl/NeonGenie`, version `3.26.0`, based on commit
+`9e7e6d388a77a9f6103fba6f982457afc873084b` before the local audit-fix commit. Record the actual installed commit
+with `git rev-parse HEAD` before installation; do not use a version string alone
+as a source identity. Root and hub packaging are distribution surfaces, not a
+claim that organizational, personal, or legacy embedded variants are identical.
+No personal version is deprecated by this change.
+
+Before switching sources, stop runtime writers, record current source/commit,
+review the destination under `${HERMES_HOME:-$HOME/.hermes}`, compare contracts,
+and retain a separate backup of outputs, sessions, and local customization.
+Do not install two different contracts with the same skill name into one profile.
+A successful compatibility check does not grant deployment/publication authority.
+
+The bundled LICENSE controls this distribution. No license grant is changed by
+these fixes; earlier grants and third-party notices remain intact.
+
+The installer validates a fresh stage on the target filesystem before replacement,
+then reads back the activated contract/version/provenance receipt. Checks use an
+isolated temporary home. Legacy `out` contents (including an out symlink) survive.
+Backups are unique, keyed by canonical destination under the active Hermes home's
+`backups///`; `.install-provenance.json` records source, base
+commit, dirty status, version, destination, and skipped checks. Stop writers during
+upgrades. Two renames have an absent-target window: this is NOT crash-atomic.
+On a reported recovery failure, retain the printed recovery directory and backup;
+do not delete it or retry blindly. Inspect the exact target and restore from the
+named backup only after validating it. No automatic source migration is implied.
+
+## Interrupted installs and stale locks
+
+SIGKILL or power loss can leave `..install-lock` beside the target.
+Locks are never automatically reclaimed: age, an empty directory, or a reused PID
+cannot prove that no installer is active. To recover:
+
+1. Stop install launchers and runtime writers. Confirm no installer is active on
+ this target (including other sessions/hosts sharing the filesystem).
+2. Inspect the exact target, sibling `.-stage-*` recovery directories
+ (especially `previous` and `failed-package`), and target-keyed backups. Retain
+ all recovery data until the original installation and outputs are accounted
+ for. Restore/validate a complete package first if activation was interrupted.
+3. Only then set `lock` to the exact path printed in the error and run
+ `rmdir -- "$lock"`. This removes only an empty lock; never use recursive
+ deletion or remove a lock whose owner/activity is uncertain. Keep launchers
+ stopped through inspection and removal to avoid races, then retry installation.
+
+Backup copies are staged in `.backup-incomplete-*` outside the target's selectable
+backup directory and published by rename after copying and writing the destination
+record. Hard termination can leave these incomplete staging trees; never select
+one for rollback. Inspect them manually after quiescing writers. Rollback skips
+legacy partial entries without a valid matching destination record.
+
+Git is optional. Archive sources, failed Git lookups, and unrelated enclosing
+worktrees record unknown Git fields as JSON `null`, never a false clean claim.
+A source-root worktree or the tracked `skills/neon-genie` hub with matching root
+contract/version supplies Git provenance. Receipts distinguish
+`source_repository_root` from `source_subdirectory` (`.` or `skills/neon-genie`).
+A dirty status lookup failure remains `null` even in a recognized worktree.
diff --git a/skills/neon-genie/scripts/install_transaction.py b/skills/neon-genie/scripts/install_transaction.py
new file mode 100644
index 0000000..c11124e
--- /dev/null
+++ b/skills/neon-genie/scripts/install_transaction.py
@@ -0,0 +1,295 @@
+"""Staged, checked upgrades with target-keyed backups and recovery.
+
+The two renames have an absent-target window; this is not crash-atomic. Stop
+runtime writers during upgrades. A retained recovery directory requires operator
+inspection. No live profile is used by validation subprocesses.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import uuid
+from pathlib import Path
+
+CHECKS = {
+ "neon-genie": [("validate_hermes_skill.py",), ("neon_genie.py", "do", "check")],
+ "sigil-forge": [("validate_hermes_skill.py",), ("sigil_forge.py", "check")],
+ "hyperlex": [("hyperlex.py", "check"), ("hyperlex.py", "smoke")],
+}
+IGNORE = shutil.ignore_patterns(
+ ".git",
+ "skills",
+ "out",
+ ".venv",
+ "__pycache__",
+ ".pytest_cache",
+ ".mypy_cache",
+ ".ruff_cache",
+ "*.pyc",
+ ".worktrees",
+ "graft",
+ ".env",
+ ".env.*",
+ ".hermes",
+ "*.pem",
+ "*.key",
+ ".superpowers",
+ "superpowers",
+)
+
+
+def install(source: Path, target: Path, kind: str, skip_checks: bool = False) -> None:
+ # Check the lexical final component before resolve follows a dangling link.
+ if any(p.is_symlink() for p in (target.absolute(), *target.absolute().parents)):
+ raise ValueError("refusing symlink target")
+ source, target = source.resolve(), target.resolve()
+ if source == target or source in target.parents or target in source.parents:
+ raise ValueError("source and target must not overlap")
+ home = Path(os.environ.get("HERMES_HOME") or str(Path.home() / ".hermes")).resolve()
+ if target in (Path("/"), Path.home().resolve(), home, home / "skills"):
+ raise ValueError("refusing installation root")
+ if target.exists() and not target.is_dir():
+ raise ValueError("target must be a directory")
+ key = hashlib.sha256(str(target).encode()).hexdigest()[:20]
+ backups = home / "backups" / kind / key
+ if target == backups or target in backups.parents or backups in target.parents:
+ raise ValueError("backup and target must not overlap")
+ # Reject payload links rather than shipping aliases into private source state.
+ for directory, dirs, files in os.walk(source, followlinks=False):
+ ignored = IGNORE(directory, dirs + files)
+ dirs[:] = [d for d in dirs if d not in ignored]
+ for name in dirs + [f for f in files if f not in ignored]:
+ if (Path(directory) / name).is_symlink():
+ raise ValueError(f"source payload symlink is not portable: {name}")
+ target.parent.mkdir(parents=True, exist_ok=True)
+ lock = target.parent / ("." + target.name + ".install-lock")
+ try:
+ lock.mkdir() # exclusive; never remove another installer's lock
+ except FileExistsError as exc:
+ raise FileExistsError(
+ f"Install lock exists: {lock}. Do not reclaim automatically. "
+ "Stop launchers and confirm no installer or runtime writer is active; "
+ "inspect the target, sibling stage/recovery directories and backups. "
+ "Only after resolving recovery, use rmdir on this exact empty lock "
+ "(never recursive removal), then retry. See references/source-and-upgrades.md."
+ ) from exc
+ workspace = None
+ retain_recovery = False
+ try:
+ workspace = Path(
+ tempfile.mkdtemp(prefix="." + kind + "-stage-", dir=target.parent)
+ )
+ stage = workspace / "package"
+ shutil.copytree(source, stage, ignore=IGNORE)
+ check_home = workspace / "check-home"
+ check_home.mkdir()
+ env = {
+ k: v
+ for k, v in os.environ.items()
+ if not k.startswith(("HYPERLEX_", "SIGIL_FORGE_"))
+ }
+ env.update(
+ HOME=str(check_home),
+ HERMES_HOME=str(check_home / ".hermes"),
+ HERMES_SKILL_DIR=str(stage),
+ SIGIL_FORGE_STATE_DIR=str(check_home / "state"),
+ PYTHONDONTWRITEBYTECODE="1",
+ )
+ if not skip_checks:
+ for command in CHECKS[kind]:
+ subprocess.run(
+ [sys.executable, str(stage / "scripts" / command[0]), *command[1:]],
+ cwd=check_home,
+ env=env,
+ check=True,
+ )
+ if (stage / "out").exists():
+ shutil.rmtree(stage / "out") # NEW staging data only
+ old_out = target / "out"
+ if old_out.is_symlink():
+ (stage / "out").symlink_to(os.readlink(old_out), target_is_directory=True)
+ elif old_out.exists():
+ shutil.copytree(old_out, stage / "out", symlinks=True)
+
+ def git(*args: str) -> str | None:
+ try:
+ r = subprocess.run(
+ check=False,
+ args=["git", "-C", str(source), *args],
+ capture_output=True,
+ text=True,
+ env={
+ k: v for k, v in os.environ.items() if not k.startswith("GIT_")
+ },
+ )
+ except OSError:
+ return None # Git is optional for archive/Python-only installs.
+ return r.stdout.strip() if r.returncode == 0 else None
+
+ # Git searches parents: accept the source root or the tracked Neon hub,
+ # not arbitrary archives nested in an unrelated worktree.
+ top = git("rev-parse", "--show-toplevel")
+ repo_root = Path(top).resolve() if top is not None else None
+ own_checkout = repo_root == source
+ subtree = "." if own_checkout else None
+ if (
+ repo_root is not None
+ and kind == "neon-genie"
+ and source == repo_root / "skills/neon-genie"
+ ):
+ tracked = git(
+ "ls-files",
+ "--error-unmatch",
+ "--",
+ "SKILL.md",
+ "VERSION",
+ str(repo_root / "SKILL.md"),
+ str(repo_root / "VERSION"),
+ )
+ try:
+ # Root/hub grants can legitimately differ; compare the contract
+ # apart from its license metadata, without changing either grant.
+ root_contract = (repo_root / "SKILL.md").read_text().splitlines()
+ hub_contract = (source / "SKILL.md").read_text().splitlines()
+ matching_root = [
+ line for line in root_contract if not line.startswith("license:")
+ ] == [
+ line for line in hub_contract if not line.startswith("license:")
+ ] and (repo_root / "VERSION").read_bytes() == (
+ source / "VERSION"
+ ).read_bytes()
+ except (OSError, UnicodeError):
+ matching_root = False
+ if tracked is not None and matching_root:
+ own_checkout = True
+ subtree = "skills/neon-genie"
+ dirty = git("status", "--porcelain") if own_checkout else None
+ receipt = {
+ "source": str(source),
+ "source_repository_root": str(repo_root) if own_checkout else None,
+ "source_subdirectory": subtree,
+ "repository": git("remote", "get-url", "origin") if own_checkout else None,
+ "source_commit": git("rev-parse", "HEAD") if own_checkout else None,
+ "source_dirty": bool(dirty) if dirty is not None else None,
+ "version": (source / "VERSION").read_text().strip(),
+ "destination": str(target),
+ "status": "UNVERIFIED" if skip_checks else "VALIDATED",
+ "checks_skipped": list(CHECKS[kind]) if skip_checks else [],
+ "validation": "staged runtime; activated contract/version read-back",
+ }
+ receipt_text = json.dumps(receipt, indent=2) + "\n"
+ (stage / ".install-provenance.json").write_text(receipt_text)
+ expected = {
+ p: (stage / p).read_bytes()
+ for p in ("SKILL.md", "VERSION", ".install-provenance.json")
+ }
+ backup = None
+ if target.exists():
+ backups.mkdir(parents=True, exist_ok=True)
+ backup = backups / uuid.uuid4().hex
+ # Incomplete copies live outside the selectable backup namespace.
+ # Publish only after the payload and destination record are complete.
+ with tempfile.TemporaryDirectory(
+ prefix=".backup-incomplete-", dir=backups.parent
+ ) as tmp:
+ pending = Path(tmp) / "backup-payload"
+ shutil.copytree(target, pending, symlinks=True)
+ marker = pending / ".backup-target.json"
+ marker.unlink(missing_ok=True)
+ marker.write_text(json.dumps({"destination": str(target)}) + "\n")
+ os.replace(pending, backup)
+ displaced = workspace / "previous"
+ # Detect a concurrent change of target identity since staging began.
+ if any(p.is_symlink() for p in (target.absolute(), *target.absolute().parents)):
+ raise ValueError("target became a symlink during staging")
+ try:
+ if target.exists():
+ os.replace(target, displaced)
+ os.replace(stage, target)
+ for relative, content in expected.items():
+ if (target / relative).read_bytes() != content:
+ raise OSError(f"activation read-back mismatch: {relative}")
+ except BaseException as original:
+ try:
+ # Infer completed renames from disk even if replace raised after
+ # its side effect. A still-staged package means target is not new.
+ if not stage.exists() and target.exists():
+ os.replace(target, workspace / "failed-package")
+ if displaced.exists():
+ os.replace(displaced, target)
+ except BaseException as recovery_error:
+ retain_recovery = True
+ raise RuntimeError(
+ f"recovery required at {workspace}; backup {backup}; "
+ f"activation: {original}; restoration: {recovery_error}"
+ ) from recovery_error
+ raise
+ print(f"Installed {kind}: {target} ({receipt['status']})")
+ if backup:
+ print(f"Backup: {backup}")
+ finally:
+ if workspace is not None and not retain_recovery:
+ shutil.rmtree(workspace)
+ lock.rmdir()
+
+
+def rollback(target: Path, kind: str, skip_checks: bool = False) -> None:
+ if any(p.is_symlink() for p in (target.absolute(), *target.absolute().parents)):
+ raise ValueError("refusing symlink target")
+ target = target.resolve()
+ home = Path(os.environ.get("HERMES_HOME") or str(Path.home() / ".hermes")).resolve()
+ key = hashlib.sha256(str(target).encode()).hexdigest()[:20]
+ backups = home / "backups" / kind / key
+ candidates = sorted(
+ backups.glob("*"), key=lambda p: p.lstat().st_mtime_ns, reverse=True
+ )
+ backup = None
+ for candidate in candidates:
+ marker = candidate / ".backup-target.json"
+ if candidate.is_symlink() or not candidate.is_dir() or marker.is_symlink():
+ continue
+ try:
+ record = json.loads(marker.read_text())
+ except (OSError, ValueError):
+ continue
+ if isinstance(record, dict) and record.get("destination") == str(target):
+ backup = candidate
+ break
+ if backup is None:
+ raise ValueError(
+ f"no target-bound backup for {target}; legacy backups require manual review"
+ )
+ # Reuse staged runtime checks, backup, activated read-back and failed-rename recovery.
+ install(backup, target, kind, skip_checks)
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("source", type=Path)
+ parser.add_argument("target")
+ parser.add_argument("kind", choices=CHECKS)
+ parser.add_argument("--skip-checks", action="store_true")
+ parser.add_argument("--rollback", action="store_true")
+ args = parser.parse_args()
+ try:
+ if not args.target.strip():
+ raise ValueError("empty target")
+ if args.rollback:
+ rollback(Path(args.target), args.kind, args.skip_checks)
+ else:
+ install(args.source, Path(args.target), args.kind, args.skip_checks)
+ except (OSError, ValueError, RuntimeError, subprocess.CalledProcessError) as exc:
+ print(f"Installation failed: {exc}", file=sys.stderr)
+ return 1
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/tests_audit/PLAN.md b/tests_audit/PLAN.md
new file mode 100644
index 0000000..b2442b9
--- /dev/null
+++ b/tests_audit/PLAN.md
@@ -0,0 +1,4 @@
+# Audit remediation
+
+User explicitly authorized C1–C6 and relevant A1/A4 fixes, local commits only.
+Plan: reproduce each defect before its fix; stage installs and validate before activation; preserve legacy output; isolate new state; repair session recipe and repository-only CI validation; align only current organizational license metadata; preserve personal variants and advisory boundaries; run sandbox regressions and existing offline checks; parent independently reviews before remote action.
diff --git a/tests_audit/test_install.py b/tests_audit/test_install.py
new file mode 100644
index 0000000..2d0b598
--- /dev/null
+++ b/tests_audit/test_install.py
@@ -0,0 +1,123 @@
+"""Installer runs only in temporary homes, never a real Hermes profile."""
+
+import os
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SKILL = "neon-genie"
+
+
+class InstallAudit(unittest.TestCase):
+ def test_profile_dry_run_has_no_writes(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ home = Path(tmp) / "home"
+ home.mkdir()
+ profile = home / "named profile"
+ env = dict(os.environ, HOME=str(home), HERMES_HOME=str(profile))
+ r = subprocess.run(
+ check=False,
+ args=["bash", str(ROOT / "install.sh"), "--dry-run"],
+ cwd=tmp,
+ env=env,
+ capture_output=True,
+ text=True,
+ )
+ self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
+ self.assertIn(str(profile / "skills" / SKILL), r.stdout)
+ self.assertEqual(list(home.iterdir()), [])
+
+ def test_failed_check_preserves_previous_install(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ home = Path(tmp) / "home"
+ home.mkdir()
+ profile = home / "profile"
+ dest = profile / "skills" / SKILL
+ dest.mkdir(parents=True)
+ (dest / "SKILL.md").write_text("previous package")
+ (dest / "out/wizard-sessions").mkdir(parents=True)
+ sentinel = dest / "out/wizard-sessions/keep.json"
+ sentinel.write_text("previous session")
+ import shutil
+
+ source = Path(tmp) / "source"
+ shutil.copytree(
+ ROOT,
+ source,
+ ignore=shutil.ignore_patterns(
+ ".git", "skills", "out", "__pycache__", ".venv"
+ ),
+ )
+ script = {
+ "neon-genie": "validate_hermes_skill.py",
+ "sigil-forge": "sigil_forge.py",
+ "hyperlex": "hyperlex.py",
+ }[SKILL]
+ (source / "scripts" / script).write_text("import sys\nsys.exit(42)\n")
+ env = dict(os.environ, HOME=str(home), HERMES_HOME=str(profile))
+ r = subprocess.run(
+ check=False,
+ args=["bash", str(source / "install.sh")],
+ cwd=tmp,
+ env=env,
+ capture_output=True,
+ text=True,
+ )
+ self.assertNotEqual(r.returncode, 0, r.stdout + r.stderr)
+ self.assertEqual((dest / "SKILL.md").read_text(), "previous package")
+ self.assertEqual(sentinel.read_text(), "previous session")
+ self.assertFalse((home / ".hermes").exists())
+
+ def test_reinstall_preserves_outputs_and_sessions(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ home = Path(tmp) / "home"
+ home.mkdir()
+ profile = home / "profile"
+ dest = profile / "skills" / SKILL
+ (dest / "out/wizard-sessions").mkdir(parents=True)
+ sentinel = dest / "out/wizard-sessions/keep.json"
+ sentinel.write_text("previous session")
+ wallpaper = dest / "out/wallpaper.png"
+ wallpaper.write_bytes(b"previous wallpaper")
+ env = dict(os.environ, HOME=str(home), HERMES_HOME=str(profile))
+ r = subprocess.run(
+ check=False,
+ args=["bash", str(ROOT / "install.sh")],
+ cwd=tmp,
+ env=env,
+ capture_output=True,
+ text=True,
+ )
+ self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
+ self.assertTrue(sentinel.is_file(), "saved session disappeared")
+ self.assertEqual(sentinel.read_text(), "previous session")
+ self.assertEqual(wallpaper.read_bytes(), b"previous wallpaper")
+ self.assertFalse((home / ".hermes").exists())
+ self.assertFalse((dest.parent / (SKILL + ".bak")).exists())
+
+ def test_profile_skills_symlink_cannot_redirect_install(self):
+ with tempfile.TemporaryDirectory() as tmp:
+ home = Path(tmp) / "home"
+ profile = home / "profile-a"
+ foreign = home / "profile-b/skills"
+ profile.mkdir(parents=True)
+ foreign.mkdir(parents=True)
+ (profile / "skills").symlink_to(foreign, target_is_directory=True)
+ env = dict(os.environ, HOME=str(home), HERMES_HOME=str(profile))
+ r = subprocess.run(
+ ["bash", str(ROOT / "install.sh")],
+ cwd=tmp,
+ env=env,
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+ self.assertNotEqual(r.returncode, 0, r.stdout + r.stderr)
+ self.assertEqual(list(foreign.iterdir()), [])
+ self.assertTrue((profile / "skills").is_symlink())
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests_audit/test_install_followup.py b/tests_audit/test_install_followup.py
new file mode 100644
index 0000000..5660e21
--- /dev/null
+++ b/tests_audit/test_install_followup.py
@@ -0,0 +1,191 @@
+"""Regression coverage for installer PR follow-up; shared across distributions."""
+
+import hashlib
+import importlib.util
+import json
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location(
+ "transaction_followup", ROOT / "scripts/install_transaction.py"
+)
+assert spec is not None and spec.loader is not None
+transaction = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(transaction)
+
+
+class FollowupTests(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.tmp.cleanup)
+ self.base = Path(self.tmp.name)
+ self.source = self.base / "source"
+ self.source.mkdir()
+ (self.source / "SKILL.md").write_text("new")
+ (self.source / "VERSION").write_text("1")
+ self.target = self.base / "target"
+ self.home = self.base / "profile"
+ env = patch.dict(os.environ, HERMES_HOME=str(self.home))
+ env.start()
+ self.addCleanup(env.stop)
+ key = hashlib.sha256(str(self.target).encode()).hexdigest()[:20]
+ self.backups = self.home / "backups/hyperlex" / key
+
+ def install(self):
+ transaction.install(self.source, self.target, "hyperlex", True)
+
+ def receipt(self):
+ return json.loads((self.target / ".install-provenance.json").read_text())
+
+ def test_archive_dirty_state_is_unknown(self):
+ self.install()
+ self.assertIsNone(self.receipt()["source_dirty"])
+
+ def test_enclosing_repository_is_not_source_provenance(self):
+ subprocess.run(["git", "init", str(self.base)], check=True, capture_output=True)
+ subprocess.run(
+ [
+ "git",
+ "-C",
+ str(self.base),
+ "-c",
+ "user.name=Test",
+ "-c",
+ "user.email=test@example.invalid",
+ "commit",
+ "--allow-empty",
+ "-m",
+ "unrelated",
+ ],
+ check=True,
+ capture_output=True,
+ )
+ subprocess.run(
+ [
+ "git",
+ "-C",
+ str(self.base),
+ "remote",
+ "add",
+ "origin",
+ "https://example.invalid/unrelated",
+ ],
+ check=True,
+ )
+ self.install()
+ for field in ("repository", "source_commit", "source_dirty"):
+ self.assertIsNone(self.receipt()[field], field)
+
+ def test_missing_git_does_not_block_install(self):
+ with patch.dict(os.environ, PATH=str(self.base / "no-executables")):
+ self.install()
+ for field in ("repository", "source_commit", "source_dirty"):
+ self.assertIsNone(self.receipt()[field], field)
+
+ def test_stale_lock_error_and_docs_are_actionable(self):
+ lock = self.target.parent / ("." + self.target.name + ".install-lock")
+ lock.mkdir()
+ with self.assertRaisesRegex(
+ FileExistsError, "Do not reclaim automatically"
+ ) as error:
+ self.install()
+ self.assertIn(str(lock), str(error.exception))
+ self.assertIn("rmdir", str(error.exception))
+ self.assertTrue(lock.is_dir())
+ self.assertFalse(self.target.exists())
+ docs = (ROOT / "references/source-and-upgrades.md").read_text()
+ for phrase in (
+ "rmdir",
+ "SIGKILL",
+ "no installer",
+ ".install-lock",
+ ".backup-incomplete-",
+ ):
+ self.assertIn(phrase, docs)
+
+ def test_tracked_neon_hub_retains_repository_and_subtree(self):
+ repo = self.base / "neon"
+ hub = repo / "skills/neon-genie"
+ hub.mkdir(parents=True)
+ for directory in (repo, hub):
+ license_id = "MIT" if directory == repo else "Apache-2.0"
+ (directory / "SKILL.md").write_text(
+ f"---\nname: neon-genie\nlicense: {license_id}\n---\n"
+ )
+ (directory / "VERSION").write_text("1")
+ subprocess.run(["git", "init", str(repo)], check=True, capture_output=True)
+ subprocess.run(["git", "-C", str(repo), "add", "."], check=True)
+ subprocess.run(
+ [
+ "git",
+ "-C",
+ str(repo),
+ "-c",
+ "user.name=Test",
+ "-c",
+ "user.email=test@example.invalid",
+ "commit",
+ "-m",
+ "hub",
+ ],
+ check=True,
+ capture_output=True,
+ )
+ transaction.install(hub, self.target, "neon-genie", True)
+ receipt = self.receipt()
+ self.assertIsNotNone(receipt["source_commit"])
+ self.assertIs(receipt["source_dirty"], False)
+ self.assertEqual(receipt["source_subdirectory"], "skills/neon-genie")
+ self.assertEqual(receipt["source_repository_root"], str(repo))
+
+ def test_invalid_utf8_enclosing_root_drops_optional_provenance(self):
+ self.test_tracked_neon_hub_retains_repository_and_subtree()
+ repo = self.base / "neon"
+ hub = repo / "skills/neon-genie"
+ (repo / "SKILL.md").write_bytes(b"\xff")
+ transaction.install(hub, self.target, "neon-genie", True)
+ receipt = self.receipt()
+ for field in (
+ "repository", "source_repository_root", "source_subdirectory",
+ "source_commit", "source_dirty",
+ ):
+ self.assertIsNone(receipt[field], field)
+ for name in ("SKILL.md", "VERSION"):
+ self.assertEqual((self.target / name).read_bytes(), (hub / name).read_bytes())
+
+ def test_partial_backup_never_published(self):
+ self.install()
+ self.install()
+ previous = set(self.backups.iterdir())
+ copytree = shutil.copytree
+
+ def interrupt(src, dst, *args, **kwargs):
+ if Path(src) == self.target:
+ Path(dst).mkdir()
+ (Path(dst) / "truncated").write_text("partial")
+ raise KeyboardInterrupt("backup copy interrupted")
+ return copytree(src, dst, *args, **kwargs)
+
+ with (
+ patch.object(transaction.shutil, "copytree", side_effect=interrupt),
+ self.assertRaises(KeyboardInterrupt),
+ ):
+ self.install()
+ self.assertEqual(set(self.backups.iterdir()), previous)
+ self.assertEqual((self.target / "SKILL.md").read_text(), "new")
+ transaction.rollback(self.target, "hyperlex", True)
+
+ def test_rollback_skips_legacy_partial_backup(self):
+ self.install()
+ self.install()
+ partial = self.backups / "partial"
+ partial.mkdir()
+ os.utime(partial, (2000000000, 2000000000))
+ transaction.rollback(self.target, "hyperlex", True)
+ self.assertEqual((self.target / "SKILL.md").read_text(), "new")
diff --git a/tests_audit/test_transaction.py b/tests_audit/test_transaction.py
new file mode 100644
index 0000000..0c7c502
--- /dev/null
+++ b/tests_audit/test_transaction.py
@@ -0,0 +1,198 @@
+"""Adversarial transactions: sandbox only, no mocked runtime successes."""
+
+import importlib.util
+import io
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location(
+ "install_transaction", ROOT / "scripts/install_transaction.py"
+)
+assert spec is not None and spec.loader is not None
+m = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(m)
+
+
+class TransactionAudit(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.tmp.cleanup)
+ self.root = Path(self.tmp.name)
+ self.source = self.root / "source"
+ self.source.mkdir()
+ (self.source / "VERSION").write_text("1")
+ (self.source / "SKILL.md").write_text("new")
+ self.target = self.root / "profile/skills/test"
+ self.env = patch.dict(
+ os.environ,
+ HOME=str(self.root / "home"),
+ HERMES_HOME=str(self.root / "profile"),
+ )
+ self.env.start()
+ self.addCleanup(self.env.stop)
+
+ def install(self):
+ m.install(self.source, self.target, "hyperlex", skip_checks=True)
+
+ def previous(self):
+ self.target.mkdir(parents=True)
+ (self.target / "SKILL.md").write_text("old")
+
+ def test_final_and_dangling_target_symlinks_rejected(self):
+ self.target.parent.mkdir(parents=True)
+ for exists in (True, False):
+ referent = self.root / str(exists)
+ if exists:
+ referent.mkdir()
+ self.target.symlink_to(referent, target_is_directory=True)
+ with self.assertRaises(ValueError):
+ self.install()
+ self.assertTrue(self.target.is_symlink())
+ self.assertFalse((referent / "SKILL.md").exists())
+ self.target.unlink()
+
+ def test_source_ancestor_rejected(self):
+ self.target = self.root
+ with self.assertRaises(ValueError):
+ self.install()
+
+ def test_secrets_not_packaged(self):
+ (self.source / ".env").write_text("FAKE_SECRET=do-not-copy")
+ (self.source / ".env.local").write_text("do-not-copy")
+ self.install()
+ self.assertFalse((self.target / ".env").exists())
+ self.assertFalse((self.target / ".env.local").exists())
+
+ def test_source_symlink_rejected(self):
+ (self.source / "linked-secret").symlink_to(self.root / "private")
+ with self.assertRaises(ValueError):
+ self.install()
+
+ def test_output_symlink_preserved_without_crawling(self):
+ self.previous()
+ private = self.root / "private"
+ private.mkdir()
+ (private / "keep").write_text("keep")
+ (self.target / "out").symlink_to(private, target_is_directory=True)
+ self.install()
+ self.assertTrue((self.target / "out").is_symlink())
+ self.assertEqual((private / "keep").read_text(), "keep")
+
+ def test_activation_failure_restores_previous(self):
+ self.previous()
+ replace = os.replace
+
+ def fail(src, dst):
+ if Path(src).name == "package":
+ raise OSError("injected activation failure")
+ return replace(src, dst)
+
+ with (
+ patch.object(m.os, "replace", side_effect=fail),
+ self.assertRaises(OSError),
+ ):
+ self.install()
+ self.assertEqual((self.target / "SKILL.md").read_text(), "old")
+
+ def test_restoration_failure_retains_recovery_tree(self):
+ self.previous()
+ replace = os.replace
+
+ def fail(src, dst):
+ if Path(src).name in ("package", "previous"):
+ raise OSError("injected rename failure")
+ return replace(src, dst)
+
+ with (
+ patch.object(m.os, "replace", side_effect=fail),
+ self.assertRaises(RuntimeError),
+ ):
+ self.install()
+ recovered = list(self.target.parent.glob(".hyperlex-stage-*/previous/SKILL.md"))
+ self.assertEqual(len(recovered), 1)
+ self.assertEqual(recovered[0].read_text(), "old")
+
+ def check_interrupted_rename(self, boundary, after, recovery):
+ self.previous()
+ replace = os.replace
+ interrupted = False
+
+ def fail(src, dst):
+ nonlocal interrupted
+ src, dst = Path(src), Path(dst)
+ if src.name == "previous":
+ if recovery == "error":
+ raise OSError("injected restoration failure")
+ if recovery == "interrupt-before":
+ raise KeyboardInterrupt("restoration before rename")
+ if recovery == "interrupt-after":
+ replace(src, dst)
+ raise KeyboardInterrupt("restoration after rename")
+ at_boundary = (
+ dst.name == "previous" if boundary == "displacement"
+ else src.name == "package"
+ )
+ if at_boundary and not interrupted:
+ interrupted = True
+ if after:
+ replace(src, dst) # Real side effect before the exception.
+ raise KeyboardInterrupt("injected installation interruption")
+ return replace(src, dst)
+
+ needs_recovery = boundary == "activation" or after
+ retained = needs_recovery and recovery != "ok"
+ with (
+ patch.object(m.os, "replace", side_effect=fail),
+ patch("sys.stdout", new_callable=io.StringIO) as output,
+ self.assertRaises(RuntimeError if retained else KeyboardInterrupt) as caught,
+ ):
+ self.install() # Explicit skip_checks=True; no fake runtime success.
+ self.assertTrue(interrupted)
+ self.assertNotIn("Installed", output.getvalue())
+ self.assertFalse((self.target.parent / ".test.install-lock").exists())
+ backups = list((self.root / "profile/backups").glob("*/*/*/SKILL.md"))
+ self.assertEqual(len(backups), 1)
+ self.assertEqual(backups[0].read_text(), "old")
+ workspaces = list(self.target.parent.glob(".*-stage-*"))
+ if retained:
+ self.assertEqual(len(workspaces), 1)
+ self.assertIn(str(workspaces[0]), str(caught.exception))
+ self.assertIn(str(backups[0].parent), str(caught.exception))
+ old = (self.target if recovery == "interrupt-after"
+ else workspaces[0] / "previous")
+ self.assertEqual((old / "SKILL.md").read_text(), "old")
+ else:
+ self.assertEqual((self.target / "SKILL.md").read_text(), "old")
+ self.assertEqual(workspaces, [])
+
+ def test_interrupted_displacement_restores_previous(self):
+ self.check_interrupted_rename("displacement", True, "ok")
+
+ def test_interrupted_rename_boundaries(self):
+ for boundary in ("displacement", "activation"):
+ for after in (False, True):
+ for recovery in ("ok", "error", "interrupt-before", "interrupt-after"):
+ with self.subTest(boundary=boundary, after=after, recovery=recovery):
+ # Each fault gets an independent real filesystem sandbox.
+ case = TransactionAudit()
+ case.setUp()
+ try:
+ case.check_interrupted_rename(boundary, after, recovery)
+ finally:
+ case.doCleanups()
+
+ def test_existing_lock_fails_closed(self):
+ self.target.parent.mkdir(parents=True)
+ lock = self.target.parent / ("." + self.target.name + ".install-lock")
+ lock.mkdir()
+ with self.assertRaises(FileExistsError):
+ self.install()
+ self.assertTrue(lock.is_dir())
+
+
+if __name__ == "__main__":
+ unittest.main()