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

skillopt_sleep/__main__.py

@@ -40,13 +40,18 @@
 from skillopt_sleep.staging import (
     StagingError,
     adopt_skills,
+    adopted_skill_names,
+    has_adopted_legacy,
     has_pending_staged_managed,
     json_safe,
+    latest_adopted_staging,
     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
 
@@ -564,12 +569,14 @@ def cmd_status(args) -> int:
             all_skills = []
             skills = []
             has_managed = False
+    revertable = latest_adopted_staging(project)
     info = {
         "night": state.night,
         "state_path": cfg.state_path,
         "project": project,
         "history_tail": state.data.get("history", [])[-5:],
         "latest_staging": latest,
+        "revertable_staging": revertable,
         "slow_memory_chars": len(state.slow_memory),
         "staged_skills": [r.get("skill_name", "") for r in skills],
         "adopted_skills": [
@@ -586,6 +593,8 @@ def cmd_status(args) -> int:
     else:
         print(f"[sleep] nights so far: {state.night}")
         print(f"[sleep] project: {project}")
+        if revertable:
+            print(f"[sleep] revertable adoption: {_display_value(revertable)}")
         if latest:
             print(f"[sleep] latest staged proposal: {_display_value(latest)}")
             if staging_error:
@@ -779,6 +788,116 @@ 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_adopted_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
+
+    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.",
+        )
+
+    if not target:
+        return fail(1, "no_staging", "[sleep] nothing to revert.")
+    if not os.path.isdir(target):
+        return fail(1, "no_staging", f"[sleep] staging directory not found: {_display_value(target)}.")
+
+    has_legacy = has_adopted_legacy(target)
+    adopted_names = adopted_skill_names(target)
+
+    if not revert_legacy and not selected and not revert_all:
+        if has_legacy and adopted_names:
+            return fail(
+                2,
+                "selection_required",
+                "[sleep] both per-skill and legacy proposals were adopted from this night; "
+                "pass --skill NAME or --all-skills, or use --legacy.",
+            )
+        if not has_legacy and not adopted_names:
+            return fail(1, "nothing_to_revert", f"[sleep] nothing to revert from {_display_value(target)}.")
+        if has_legacy:
+            revert_legacy = True
+        else:
+            revert_all = True
+
+    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
+
+    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
+
+
 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 +991,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 +1039,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":