feat: audit loop skill installations

skills/loop-engineering/SKILL.md

@@ -12,8 +12,9 @@ not a loop design.
 
 1. Skip this skill when one action plus one check is sufficient.
 2. For interactive, resumable, or delegated loops, use the state script below.
-3. For scheduled loops, also read [references/hosts.md](references/hosts.md) and
-   require a real, authorized recurrence primitive.
+3. For scheduled loops or installation drift, also read
+   [references/hosts.md](references/hosts.md); require a real recurrence
+   primitive for scheduling and use its audit command for duplicate copies.
 4. For exact transition, effect, worklog, or handoff rules, read
    [references/protocol.md](references/protocol.md).
 

skills/loop-engineering/references/hosts.md

@@ -16,6 +16,9 @@ tracking.
 - After intervention or a scheduled wakeup, the agent creates a successor state
   bound to the terminal predecessor and replays the stopping check before
   continuing. Treat supplied intervention as pending until that check passes.
+- For duplicate installations, run `scripts/install_audit.py --canonical
+  <skill-dir>`. `--link-identical` replaces only byte-identical directories;
+  any divergent root fails the whole preflight before writes.
 
 ## Codex
 

skills/loop-engineering/scripts/install_audit.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+"""Audit and safely consolidate loop-engineering skill installations."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import pathlib
+import shutil
+import sys
+import uuid
+
+
+CLEAN_STATUSES = {"absent", "linked", "source"}
+DIVERGENT_STATUSES = {
+    "broken-symlink",
+    "divergent-copy",
+    "divergent-symlink",
+    "unsupported",
+}
+
+
+def tree_digest(root: pathlib.Path) -> str:
+    digest = hashlib.sha256()
+    for path in sorted(root.rglob("*"), key=lambda item: item.as_posix()):
+        relative = path.relative_to(root).as_posix().encode()
+        if path.is_symlink():
+            digest.update(
+                b"L\0" + relative + b"\0" + os.readlink(path).encode() + b"\0"
+            )
+        elif path.is_file():
+            digest.update(b"F\0" + relative + b"\0")
+            digest.update(hashlib.sha256(path.read_bytes()).digest())
+        elif path.is_dir():
+            digest.update(b"D\0" + relative + b"\0")
+    return digest.hexdigest()
+
+
+def same_path(left: pathlib.Path, right: pathlib.Path) -> bool:
+    try:
+        return left.resolve(strict=True) == right.resolve(strict=True)
+    except FileNotFoundError:
+        return False
+
+
+def classify(
+    root: pathlib.Path,
+    canonical: pathlib.Path,
+    canonical_digest: str,
+) -> dict[str, str]:
+    result = {"path": str(root)}
+    if root.is_symlink():
+        try:
+            resolved = root.resolve(strict=True)
+        except FileNotFoundError:
+            return result | {"status": "broken-symlink", "target": os.readlink(root)}
+        status = "linked" if same_path(resolved, canonical) else "divergent-symlink"
+        return result | {"status": status, "target": str(resolved)}
+    if not root.exists():
+        return result | {"status": "absent"}
+    if not root.is_dir():
+        return result | {"status": "unsupported"}
+    if same_path(root, canonical):
+        return result | {"status": "source", "digest": canonical_digest}
+    digest = tree_digest(root)
+    status = "duplicate-identical" if digest == canonical_digest else "divergent-copy"
+    return result | {"status": status, "digest": digest}
+
+
+def replace_identical_copy(root: pathlib.Path, canonical: pathlib.Path) -> None:
+    token = uuid.uuid4().hex
+    backup = root.with_name(f"{root.name}.backup-{token}")
+    temporary_link = root.with_name(f"{root.name}.link-{token}")
+    os.symlink(f"{canonical}{os.sep}", temporary_link, target_is_directory=True)
+    root.rename(backup)
+    try:
+        os.replace(temporary_link, root)
+    except Exception:
+        backup.rename(root)
+        temporary_link.unlink(missing_ok=True)
+        raise
+    shutil.rmtree(backup)
+
+
+def default_roots(home: pathlib.Path) -> list[pathlib.Path]:
+    return [
+        home / ".codex/skills/loop-engineering",
+        home / ".agents/skills/loop-engineering",
+        home / ".claude/skills/loop-engineering",
+        home / ".cursor/skills/loop-engineering",
+    ]
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Audit loop-engineering installations against one canonical source."
+    )
+    parser.add_argument(
+        "--canonical",
+        type=pathlib.Path,
+        default=pathlib.Path(__file__).parents[1],
+        help="canonical loop-engineering directory (defaults to this script's skill)",
+    )
+    parser.add_argument(
+        "--root",
+        action="append",
+        type=pathlib.Path,
+        help="installation root to audit; repeat to override the four user defaults",
+    )
+    parser.add_argument(
+        "--link-identical",
+        action="store_true",
+        help="replace byte-identical copied directories with canonical symlinks",
+    )
+    parser.add_argument("--json", action="store_true", help="emit JSON")
+    return parser
+
+
+def render(entries: list[dict[str, str]], as_json: bool) -> None:
+    if as_json:
+        print(json.dumps(entries, indent=2, sort_keys=True))
+        return
+    for entry in entries:
+        detail = entry.get("target") or entry.get("digest", "")
+        suffix = f" ({detail})" if detail else ""
+        print(f"{entry['status']}: {entry['path']}{suffix}")
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    canonical = args.canonical.expanduser().resolve(strict=True)
+    if not (canonical / "SKILL.md").is_file():
+        print(
+            f"install-audit: canonical skill is missing SKILL.md: {canonical}",
+            file=sys.stderr,
+        )
+        return 2
+
+    roots = args.root or default_roots(pathlib.Path.home())
+    roots = [root.expanduser().absolute() for root in roots]
+    canonical_digest = tree_digest(canonical)
+    entries = [classify(root, canonical, canonical_digest) for root in roots]
+
+    if args.link_identical:
+        if any(entry["status"] in DIVERGENT_STATUSES for entry in entries):
+            render(entries, args.json)
+            print(
+                "install-audit: refusing all writes while a divergent installation exists",
+                file=sys.stderr,
+            )
+            return 1
+        for root, entry in zip(roots, entries):
+            if entry["status"] == "duplicate-identical":
+                replace_identical_copy(root, canonical)
+        entries = [classify(root, canonical, canonical_digest) for root in roots]
+
+    render(entries, args.json)
+    return 0 if all(entry["status"] in CLEAN_STATUSES for entry in entries) else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

skills/loop-engineering/tests/test_install_audit.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+"""Fixtures for deterministic loop-engineering installation audits."""
+
+from __future__ import annotations
+
+import json
+import pathlib
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+SCRIPT = pathlib.Path(__file__).parents[1] / "scripts" / "install_audit.py"
+
+
+class InstallAuditTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary_directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary_directory.cleanup)
+        self.root = pathlib.Path(self.temporary_directory.name)
+        self.canonical = self.root / "canonical"
+        self.canonical.mkdir()
+        (self.canonical / "SKILL.md").write_text("canonical\n")
+        (self.canonical / "scripts").mkdir()
+        (self.canonical / "scripts/tool.py").write_text("print('ok')\n")
+
+    def run_cli(
+        self,
+        *roots: pathlib.Path,
+        link_identical: bool = False,
+        expected_returncode: int = 0,
+    ) -> tuple[subprocess.CompletedProcess[str], list[dict[str, str]]]:
+        arguments = [
+            sys.executable,
+            str(SCRIPT),
+            "--canonical",
+            str(self.canonical),
+            "--json",
+        ]
+        for root in roots:
+            arguments.extend(("--root", str(root)))
+        if link_identical:
+            arguments.append("--link-identical")
+        result = subprocess.run(arguments, capture_output=True, text=True, check=False)
+        self.assertEqual(
+            result.returncode,
+            expected_returncode,
+            msg=f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}",
+        )
+        return result, json.loads(result.stdout)
+
+    def make_copy(self, name: str) -> pathlib.Path:
+        destination = self.root / name
+        shutil.copytree(self.canonical, destination)
+        return destination
+
+    def test_audit_classifies_source_link_copy_and_absence(self) -> None:
+        linked = self.root / "linked"
+        linked.symlink_to(self.canonical, target_is_directory=True)
+        copied = self.make_copy("copied")
+        absent = self.root / "absent"
+
+        _, entries = self.run_cli(
+            self.canonical,
+            linked,
+            copied,
+            absent,
+            expected_returncode=1,
+        )
+
+        self.assertEqual(
+            [entry["status"] for entry in entries],
+            ["source", "linked", "duplicate-identical", "absent"],
+        )
+
+    def test_repair_is_fail_closed_when_any_copy_diverges(self) -> None:
+        identical = self.make_copy("identical")
+        divergent = self.make_copy("divergent")
+        (divergent / ".DS_Store").write_text("unexpected\n")
+
+        result, entries = self.run_cli(
+            identical,
+            divergent,
+            link_identical=True,
+            expected_returncode=1,
+        )
+
+        self.assertIn("refusing all writes", result.stderr)
+        self.assertEqual(
+            [entry["status"] for entry in entries],
+            ["duplicate-identical", "divergent-copy"],
+        )
+        self.assertTrue(identical.is_dir())
+        self.assertFalse(identical.is_symlink())
+
+    def test_repair_replaces_only_identical_copy_with_symlink(self) -> None:
+        identical = self.make_copy("identical")
+
+        _, entries = self.run_cli(identical, link_identical=True)
+
+        self.assertEqual(entries[0]["status"], "linked")
+        self.assertTrue(identical.is_symlink())
+        self.assertEqual(identical.resolve(), self.canonical.resolve())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/run.sh

@@ -141,6 +141,7 @@ examples = (skill / "references/examples.md").read_text()
 hosts = (skill / "references/hosts.md").read_text()
 protocol = (skill / "references/protocol.md").read_text()
 state_script = (skill / "scripts/loop_state.py").read_text()
+install_script = (skill / "scripts/install_audit.py").read_text()
 references = {path.name for path in (skill / "references").glob("*.md")}
 checks = {
     "thin portable root": len(root.splitlines()) <= 80,
@@ -179,6 +180,12 @@ checks = {
         and "Malformed CLI usage exits `2`" in protocol
         and "state contract exits `3`" in protocol
     ),
+    "fail-closed installation consolidation": (
+        "install_audit.py --canonical" in hosts
+        and "--link-identical" in hosts
+        and "refusing all writes" in install_script
+        and "divergent-copy" in install_script
+    ),
     "atomic state write": "os.replace" in state_script,
     "host differences deferred": references == {"examples.md", "hosts.md", "protocol.md"},
     "cross-host continuation": (
@@ -292,7 +299,9 @@ PY
     fail "worklog PR reconciliation fixtures"
   fi
 
-  if python3 -m unittest skills/loop-engineering/tests/test_loop_state.py >/dev/null; then
+  if python3 -m unittest \
+    skills/loop-engineering/tests/test_loop_state.py \
+    skills/loop-engineering/tests/test_install_audit.py >/dev/null; then
     ok "loop-engineering state transition fixtures"
   else
     fail "loop-engineering state transition fixtures"