feat(sleep): add revert command to rollback adopted proposals

skillopt_sleep/__main__.py

@@ -44,9 +44,11 @@
     json_safe,
     latest_staging,
     pending_staged_skills,
+    revert_skills,
     staged_skills,
 )
 from skillopt_sleep.staging import adopt as adopt_staging
+from skillopt_sleep.staging import revert as revert_staging
 from skillopt_sleep.state import SleepState
 from skillopt_sleep.tasks_file import load_tasks_file, make_tasks_payload, write_tasks_file
 
@@ -779,6 +781,125 @@ def fail(code: int, kind: str, message: str, **extra: Any) -> int:
     return 0
 
 
+def cmd_revert(args) -> int:
+    cfg = _cfg_from_args(args)
+    project = cfg.get("invoked_project") or os.getcwd()
+    target = args.staging or latest_staging(project)
+
+    def fail(code: int, kind: str, message: str, **extra: Any) -> int:
+        safe_message = _display_value(message)
+        if args.json:
+            payload = {
+                "ok": False,
+                "error": kind,
+                "message": safe_message,
+                "staging_dir": _display_value(target or ""),
+            }
+            payload.update(_redact_deep(extra))
+            print(json.dumps(payload, ensure_ascii=False, indent=2))
+        else:
+            print(safe_message)
+        return code
+
+    if not target or not os.path.isdir(target):
+        return fail(1, "no_staging", "[sleep] nothing to revert (no staging dir).")
+    raw_selected = list(getattr(args, "skills", None) or [])
+    if any(not str(name).strip() for name in raw_selected):
+        return fail(
+            2,
+            "invalid_selection",
+            "[sleep] --skill names must be non-empty.",
+        )
+    selected = [str(name).strip() for name in raw_selected]
+    revert_all = bool(getattr(args, "all_skills", False))
+    revert_legacy = bool(getattr(args, "legacy", False))
+    if sum((bool(selected), revert_all, revert_legacy)) > 1:
+        return fail(
+            2,
+            "invalid_selection",
+            "[sleep] use exactly one of --skill, --all-skills, or --legacy.",
+        )
+    try:
+        # Reusing staged_skills here as a sanity check that it's a valid staging dir.
+        rows = staged_skills(target)
+    except Exception as exc:
+        return fail(
+            1,
+            "invalid_staging",
+            f"[sleep] cannot read staged skills for revert: {exc}",
+        )
+    if revert_legacy:
+        try:
+            updated = revert_staging(target)
+        except StagingError as exc:
+            return fail(2, "revert_refused", f"[sleep] revert refused: {exc}")
+        except (OSError, ValueError, KeyError, json.JSONDecodeError) as exc:
+            return fail(1, "revert_failed", f"[sleep] revert failed: {exc}")
+        if args.json:
+            print(json.dumps({
+                "ok": True,
+                "staging_dir": target,
+                "mode": "legacy",
+                "reverted_skills": [],
+                "updated_paths": updated,
+            }, ensure_ascii=False, indent=2))
+        else:
+            print(f"[sleep] reverted managed proposal from {_display_value(target)}")
+            for path in updated:
+                print(f"   <- {_display_value(path)}")
+            if not updated:
+                print("[sleep] (no adopted managed changes to revert)")
+        return 0
+    if selected or revert_all:
+        # Assuming revert_skills behaves similarly to adopt_skills returning receipts.
+        names = selected if not revert_all else None
+        try:
+            receipts = revert_skills(target, names)
+        except StagingError as exc:
+            return fail(2, "revert_refused", f"[sleep] revert refused: {exc}")
+        except OSError as exc:
+            return fail(1, "revert_failed", f"[sleep] revert failed: {exc}")
+        if args.json:
+            print(json.dumps(json_safe({
+                "ok": True,
+                "staging_dir": target,
+                "mode": "skills",
+                "reverted_skills": [receipt.__dict__ for receipt in receipts],
+                "updated_paths": [receipt.live_skill_path for receipt in receipts],
+            }), ensure_ascii=False, indent=2))
+        else:
+            print(f"[sleep] reverted from {_display_value(target)}")
+            for receipt in receipts:
+                print(
+                    f"   <- {_display_value(receipt.skill_name)}: "
+                    f"{_display_value(receipt.live_skill_path)}"
+                )
+            if not receipts:
+                print("[sleep] (no skills to revert in the selection)")
+        return 0
+    try:
+        updated = revert_staging(target)
+    except StagingError as exc:
+        return fail(2, "revert_refused", f"[sleep] revert refused: {exc}")
+    except (OSError, ValueError, KeyError, json.JSONDecodeError) as exc:
+        return fail(1, "revert_failed", f"[sleep] revert failed: {exc}")
+    if args.json:
+        print(json.dumps({
+            "ok": True,
+            "staging_dir": target,
+            "mode": "legacy",
+            "reverted_skills": [],
+            "updated_paths": updated,
+        }, ensure_ascii=False, indent=2))
+    else:
+        print(f"[sleep] reverted from {_display_value(target)}")
+        for path in updated:
+            print(f"   <- {_display_value(path)}")
+        if not updated:
+            print("[sleep] (no adopted changes to revert)")
+    return 0
+
+
 def cmd_harvest(args) -> int:
     cfg = _cfg_from_args(args)
     session_limit = cfg.get("max_sessions_per_night", 0) or cfg.get("max_tasks_per_night", 40) * 3
@@ -872,6 +993,21 @@ def main(argv=None) -> int:
         "--legacy", action="store_true",
         help="adopt only the staged managed skill/memory proposal",
     )
+    p_revert = sub.add_parser("revert", help="revert an adopted proposal using its backups")
+    _add_common(p_revert)
+    p_revert.add_argument("--staging", default="", help="specific staging dir")
+    p_revert.add_argument(
+        "--skill", action="append", default=[], dest="skills",
+        help="revert this adopted skill (repeatable)",
+    )
+    p_revert.add_argument(
+        "--all-skills", action="store_true", dest="all_skills",
+        help="revert every adopted per-skill proposal",
+    )
+    p_revert.add_argument(
+        "--legacy", action="store_true",
+        help="revert only the adopted managed skill/memory proposal",
+    )
     p_harvest = sub.add_parser("harvest", help="debug: show mined tasks")
     _add_common(p_harvest)
     p_harvest.add_argument("--output", default="", help="write mined tasks JSON for review")
@@ -905,6 +1041,8 @@ def main(argv=None) -> int:
         return cmd_status(args)
     if args.cmd == "adopt":
         return cmd_adopt(args)
+    if args.cmd == "revert":
+        return cmd_revert(args)
     if args.cmd == "harvest":
         return cmd_harvest(args)
     if args.cmd == "schedule":

skillopt_sleep/staging.py

@@ -3152,3 +3152,130 @@ def adopt(staging_dir: str) -> List[str]:
             receipt_after=receipt_after,
         )
         return updated
+
+
+def revert_skills(
+    staging_dir: str, skill_names: Optional[Sequence[str]] = None
+) -> List[AdoptedSkill]:
+    """Revert adopted skills back to their pre-adoption state using backups."""
+    staging_dir = _canonical_staging_dir(staging_dir)
+    receipt_path = os.path.join(staging_dir, "adopted_skills.json")
+    if not os.path.exists(receipt_path):
+        return []
+    with open(receipt_path, encoding="utf-8") as f:
+        existing_receipts = json.load(f)
+
+    if skill_names is not None:
+        names = {str(name).strip() for name in skill_names if str(name).strip()}
+        rows = [r for r in existing_receipts if r["skill_name"] in names]
+    else:
+        rows = list(existing_receipts)
+
+    if not rows:
+        return []
+
+    receipts_to_return: List[AdoptedSkill] = []
+    remaining_receipts: List[Dict[str, Any]] = [
+        r for r in existing_receipts if r not in rows
+    ]
+
+    for row in rows:
+        name = row["skill_name"]
+        live = row["live_skill_path"]
+        before_sha = row["sha256_before"]
+        after_sha = row["sha256_after"]
+        backup = row["backup_path"]
+
+        current, current_mode, _ = _file_snapshot(live)
+        current_sha = _bytes_sha256(current)
+        if current_sha == before_sha:
+            receipts_to_return.append(AdoptedSkill(
+                skill_name=name,
+                live_skill_path=live,
+                sha256_before=after_sha,
+                sha256_after=before_sha,
+                backup_path="",
+            ))
+            continue
+
+        if current_sha != after_sha:
+            raise StagingError(f"live skill for {name!r} changed since adoption; cannot safely revert")
+
+        if before_sha != "":
+            if not backup or not os.path.lexists(backup):
+                raise StagingError(f"backup for {name!r} is missing; cannot revert")
+            with open(backup, "rb") as f:
+                proposed_bytes = f.read()
+            if hashlib.sha256(proposed_bytes).hexdigest() != before_sha:
+                raise StagingError(f"backup for {name!r} changed; cannot safely revert")
+            _write_atomic_bytes(live, proposed_bytes, mode=current_mode)
+        else:
+            if current is not None:
+                os.unlink(live)
+
+        receipts_to_return.append(AdoptedSkill(
+            skill_name=name,
+            live_skill_path=live,
+            sha256_before=after_sha,
+            sha256_after=before_sha,
+            backup_path="",
+        ))
+
+    _write_atomic(
+        receipt_path,
+        json.dumps(remaining_receipts, ensure_ascii=False, indent=2),
+        create_parents=False,
+    )
+    return receipts_to_return
+
+
+def revert(staging_dir: str) -> List[str]:
+    """Revert adopted legacy managed skills back to their pre-adoption state."""
+    staging_dir = _canonical_staging_dir(staging_dir)
+    receipt_path = os.path.join(staging_dir, "adopted_legacy.json")
+    if not os.path.exists(receipt_path):
+        return []
+    with open(receipt_path, encoding="utf-8") as f:
+        existing_receipts = json.load(f)
+
+    if not existing_receipts:
+        return []
+
+    updated: List[str] = []
+
+    for row in existing_receipts:
+        label = row["target"]
+        live = row["live_path"]
+        before_sha = row["sha256_before"]
+        after_sha = row["sha256_after"]
+        backup = row["backup_path"]
+
+        current, current_mode, _ = _file_snapshot(live)
+        current_sha = _bytes_sha256(current)
+        if current_sha == before_sha:
+            updated.append(live)
+            continue
+
+        if current_sha != after_sha:
+            raise StagingError(f"legacy {label} changed since adoption; cannot safely revert")
+
+        if before_sha != "":
+            if not backup or not os.path.lexists(backup):
+                raise StagingError(f"legacy backup for {label} is missing; cannot revert")
+            with open(backup, "rb") as f:
+                proposed_bytes = f.read()
+            if hashlib.sha256(proposed_bytes).hexdigest() != before_sha:
+                raise StagingError(f"legacy backup for {label} changed; cannot safely revert")
+            _write_atomic_bytes(live, proposed_bytes, mode=current_mode)
+        else:
+            if current is not None:
+                os.unlink(live)
+
+        updated.append(live)
+
+    _write_atomic(
+        receipt_path,
+        json.dumps([], ensure_ascii=False, indent=2),
+        create_parents=False,
+    )
+    return updated

tests/test_sleep_revert.py

@@ -0,0 +1,99 @@
+"""Tests for reverting adopted skills."""
+import hashlib
+import json
+import os
+import shutil
+import tempfile
+import unittest
+
+from skillopt_sleep.staging import (
+    SkillProposal,
+    adopt,
+    adopt_skills,
+    revert,
+    revert_skills,
+    write_staging,
+    StagingError,
+)
+from skillopt_sleep.types import SleepReport
+
+
+def _sha(text):
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+def _write(path, text):
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w", encoding="utf-8") as f:
+        f.write(text)
+
+class TestRevert(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.project = os.path.join(self.tmp, "project")
+        self.staging = os.path.join(self.tmp, "staging")
+        os.makedirs(self.project)
+        os.makedirs(self.staging)
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def test_revert_skills(self):
+        skill_live = os.path.join(self.project, "skills", "hello", "SKILL.md")
+        _write(skill_live, "before")
+        before_sha = _sha("before")
+
+        proposal = SkillProposal(
+            skill_name="hello",
+            proposed_skill="after",
+            live_skill_path=skill_live,
+        )
+        report = SleepReport(night=1, project=self.project)
+        self.staging = write_staging(
+            self.tmp,
+            report=report,
+            proposed_skill=None, proposed_memory=None,
+            live_skill_path=None, live_memory_path=None,
+            skill_proposals=[proposal],
+            report_md="",
+        )
+
+        receipts = adopt_skills(self.staging)
+        self.assertEqual(len(receipts), 1)
+
+        with open(skill_live, "r", encoding="utf-8") as f:
+            self.assertEqual(f.read(), "after")
+
+        reverted = revert_skills(self.staging)
+        self.assertEqual(len(reverted), 1)
+
+        with open(skill_live, "r", encoding="utf-8") as f:
+            self.assertEqual(f.read(), "before")
+
+    def test_revert_legacy(self):
+        skill_live = os.path.join(self.project, "SKILL.md")
+        _write(skill_live, "legacy before")
+        before_sha = _sha("legacy before")
+
+        report = SleepReport(night=1, project=self.project, accepted=True)
+        self.staging = write_staging(
+            self.tmp,
+            report=report,
+            proposed_skill="legacy after",
+            proposed_memory=None,
+            live_skill_path=skill_live,
+            live_memory_path=None,
+            skill_proposals=[],
+            report_md="",
+        )
+
+        updated = adopt(self.staging)
+        self.assertEqual(len(updated), 1)
+
+        with open(skill_live, "r", encoding="utf-8") as f:
+            self.assertEqual(f.read(), "legacy after")
+
+        reverted = revert(self.staging)
+        self.assertEqual(len(reverted), 1)
+
+        with open(skill_live, "r", encoding="utf-8") as f:
+            self.assertEqual(f.read(), "legacy before")