diff --git a/skillopt_sleep/__main__.py b/skillopt_sleep/__main__.py index e3b7794c..08e2210a 100644 --- a/skillopt_sleep/__main__.py +++ b/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": diff --git a/skillopt_sleep/staging.py b/skillopt_sleep/staging.py index 615c5ac8..5f96fc32 100644 --- a/skillopt_sleep/staging.py +++ b/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 diff --git a/tests/test_sleep_revert.py b/tests/test_sleep_revert.py new file mode 100644 index 00000000..786dd91f --- /dev/null +++ b/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")