diff --git a/CHANGELOG.md b/CHANGELOG.md index 513840a9..418fbe62 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,15 @@ All notable changes to SkillOpt are documented here. This project adheres to ## [Unreleased] ### Added +- **`skillopt-sleep revert`**, undoing an adoption by reversing its receipt. + A live document adoption replaced is restored from its immutable backup; one + adoption created is removed, since that is the state being returned to. + Selection mirrors `adopt` (`--skill`, `--all-skills`, `--legacy`, + `--staging`), and a bare `revert` targets the most recent night with an + adoption still on record. Reverting consumes the backup and clears the night's + receipt rows, so the night can be adopted again. It refuses when the live file + no longer matches what adoption wrote — it was edited since, and restoring the + backup would discard that work — or when a backup is missing or fails its pin. - **SkillOpt-Sleep multi-skill fan-out and reviewed subset adoption**: each hinted skill is consolidated from its own pinned live baseline, staged as an independent proposal with per-skill gate evidence, and promoted only through diff --git a/docs/reference/cli.md b/docs/reference/cli.md index f0ea40a5..52b5f3d8 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -123,7 +123,7 @@ skillopt-sleep [options] python -m skillopt_sleep [options] ``` -Actions are `run`, `dry-run`, `status`, `adopt`, `harvest`, `schedule`, and +Actions are `run`, `dry-run`, `status`, `adopt`, `revert`, `harvest`, `schedule`, and `unschedule`. Common options include: | Argument | Description | @@ -169,6 +169,42 @@ mining, replay, judging, and reflection prompts derived from harvested transcripts and tasks to its selected provider. Review that provider's data-retention and privacy policy before processing sensitive sessions. +### Reverting an adoption + +`revert` undoes an adoption, restoring the files it replaced: + +```bash +skillopt-sleep revert # undo the last adoption for this project +skillopt-sleep revert --legacy # undo only the managed pair +skillopt-sleep revert --skill NAME # undo one adopted skill (repeatable) +skillopt-sleep revert --all-skills # undo every adopted per-skill proposal +skillopt-sleep revert --staging # undo a specific night +``` + +It is defined against the adoption receipts, so it reverses exactly what +adoption recorded. A live document that adoption **replaced** is restored from +its immutable backup; one that adoption **created** is removed, because that is +the state being returned to. Reverting consumes the backup and clears the +night's receipt rows, which returns the staging directory to its pre-adopt shape +so it can be adopted again. + +Without a selection, `revert` acts on whichever kind of adoption the night +holds; when it holds both, it lists the adopted skills and exits so the choice +is explicit. Without `--staging`, it targets the most recent night with an +adoption still on record — deliberately not the newest staged night, whose +proposal may never have been adopted. `status` reports that directory as +`revertable_staging`. + +`revert` refuses rather than reporting a successful no-op when the live file no +longer matches what adoption wrote — it was edited or replaced since, and +restoring the backup would discard that work — or when the backup is missing or +fails its pin. Directories that adoption created are left in place: the receipt +does not record which ones it made, and removing a directory whose ownership was +not durably recorded is the same fail-closed case adoption's own recovery path +refuses. Revert is not WAL-journaled the way adoption is; every step is checked +against the receipt's pins and converges, so an interrupted revert is completed +by running it again. + ### VS Code GitHub Copilot Chat source `--source copilot` reads local VS Code GitHub Copilot Chat session logs from diff --git a/docs/sleep/README.md b/docs/sleep/README.md index 2576c127..40c53d9e 100644 --- a/docs/sleep/README.md +++ b/docs/sleep/README.md @@ -103,9 +103,19 @@ skillopt-sleep status # show state + the latest staged proposal skillopt-sleep adopt --legacy # apply a reviewed managed proposal skillopt-sleep adopt --skill NAME # adopt one staged skill (repeatable) skillopt-sleep adopt --all-skills # adopt every still-pending fan-out skill +skillopt-sleep revert # undo the last adoption, restoring what it replaced skillopt-sleep schedule # install a nightly cron entry for this project ``` +> **Adoption is reversible.** `revert` reverses an adoption receipt: a document +> adoption replaced is restored from its immutable backup, one adoption created +> is removed. It matters most when `--auto-adopt` runs unattended from the +> nightly schedule — the gate is a held-out validation gate, not an oracle, so a +> night can accept an edit that scores better on a handful of mined tasks and +> still be worse in daily use. It refuses rather than silently discarding work +> if you edited the live file after adopting. See +> [Reverting an adoption](../reference/cli.md#reverting-an-adoption). + > **Version note.** This page tracks `main`. PyPI 0.2.0 provides the base > commands above. Cursor source/backend/plugin support, VS Code Copilot > transcript harvesting, Pi source/backend support, Sleep handoff, non-Azure diff --git a/skillopt_sleep/__main__.py b/skillopt_sleep/__main__.py index 6875ad21..10b5b98a 100644 --- a/skillopt_sleep/__main__.py +++ b/skillopt_sleep/__main__.py @@ -5,6 +5,7 @@ python -m skillopt_sleep status # show state + latest staged proposal python -m skillopt_sleep adopt # apply the latest staged proposal (with backup) python -m skillopt_sleep adopt --skill NAME # adopt one staged skill (repeatable) + python -m skillopt_sleep revert # undo the last adoption, restoring backups python -m skillopt_sleep harvest # just print what would be mined (debug) Common flags: @@ -40,13 +41,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 @@ -550,6 +556,10 @@ def cmd_status(args) -> int: state = SleepState.load(cfg.state_path) project = cfg.get("invoked_project") or os.getcwd() latest = latest_staging(project) + try: + revertable = latest_adopted_staging(project) + except (OSError, StagingError): + revertable = None skills = [] all_skills = [] has_managed = False @@ -578,6 +588,9 @@ def cmd_status(args) -> int: if row not in skills ], "has_managed_proposal": has_managed, + # What a bare `revert` would undo. May be an older night than + # `latest_staging`, whose proposal need never have been adopted. + "revertable_staging": revertable, } if staging_error: info["staging_error"] = staging_error @@ -586,6 +599,9 @@ 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)}") + print("[sleep] undo it with: skillopt-sleep revert") if latest: print(f"[sleep] latest staged proposal: {_display_value(latest)}") if staging_error: @@ -779,6 +795,110 @@ 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() + # Default to the last *adopted* night, not the newest staged one: reverting + # a proposal that was never adopted would restore a backup for a change the + # live files never received. + 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 + + if not target or not os.path.isdir(target): + return fail( + 1, "no_adoption", + "[sleep] nothing to revert (no adopted proposal for this project).", + ) + 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: + adopted_skills = adopted_skill_names(target) + adopted_legacy = has_adopted_legacy(target) + except (OSError, StagingError) as exc: + return fail(1, "invalid_staging", f"[sleep] cannot read adoption receipts: {exc}") + + if not adopted_skills and not adopted_legacy: + return fail( + 2, "not_adopted", + f"[sleep] nothing from {_display_value(target)} is adopted; " + "there is nothing to revert.", + ) + # With only one kind of adoption on record the intent is unambiguous, so an + # explicit selection is required only when both are present. + if not selected and not revert_all and not revert_legacy: + if adopted_skills and adopted_legacy: + message = ( + "[sleep] this night has both managed and per-skill adoptions; " + "pass --skill NAME, --all-skills, or --legacy." + ) + if args.json: + return fail(2, "selection_required", message, + adopted_skills=adopted_skills) + print(_display_value(message)) + for name in adopted_skills: + print(f" {_display_value(name)!r}") + return 2 + revert_legacy = adopted_legacy + revert_all = bool(adopted_skills) + + try: + if revert_legacy: + results = revert_staging(target) + mode = "legacy" + else: + names = adopted_skills if revert_all else selected + results = revert_skills(target, names) + mode = "skills" + 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": mode, + "reverted": [result.__dict__ for result in results], + }), ensure_ascii=False, indent=2)) + else: + print(f"[sleep] reverted {_display_value(target)}") + for result in results: + live = _display_value(result.live_path) + if result.already_reverted: + print(f" unchanged -> {live} (already reverted)") + elif result.removed: + print(f" removed -> {live} (adopt had created it)") + else: + print(f" restored -> {live}") + if not results: + print("[sleep] (nothing in the selection was adopted)") + 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 +992,23 @@ 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="undo an adoption, restoring the pre-adopt files") + _add_common(p_revert) + p_revert.add_argument("--staging", default="", + help="specific staging dir (default: last adopted)") + 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") @@ -892,6 +1029,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 e2ecfbd5..67478e71 100644 --- a/skillopt_sleep/staging.py +++ b/skillopt_sleep/staging.py @@ -3100,3 +3100,242 @@ def adopt(staging_dir: str) -> List[str]: receipt_after=receipt_after, ) return updated + + +# ── reverting an adoption ──────────────────────────────────────────────────── +# `adopt` publishes a receipt (adopted_legacy.json / adopted_skills.json) that +# already pins everything an undo needs: the live path, the sha256 it held +# before adoption ("" when no file existed), the sha256 adoption wrote, and the +# immutable backup. `revert` reads that receipt and reverses it. +# +# Unlike adoption, this is not WAL-journaled. It does not need to be: every step +# is verified against the receipt's pins and converges, so a revert interrupted +# part-way is completed by running it again. Promoting it into the durable +# transaction is possible but would mean teaching `_TransactionTarget` to delete +# a path, which the adoption WAL schema has no representation for today. + +_LEGACY_RECEIPT_FILE = "adopted_legacy.json" +_SKILLS_RECEIPT_FILE = "adopted_skills.json" + + +@dataclass +class RevertedTarget: + """Receipt for one reverted path: what it went back to.""" + + key: str # "skill"/"memory" for the legacy pair, else skill name + live_path: str + sha256_restored: str # "" when the path was removed instead of restored + removed: bool # adopt had created this file, so revert deleted it + already_reverted: bool # live already matched the pre-adopt pin; nothing to do + + +def _receipt_rows(staging_dir: str, filename: str) -> List[Dict[str, Any]]: + payload, _original, _mode, _file_id = _read_receipt_file( + os.path.join(staging_dir, filename) + ) + return payload + + +def _publish_remaining_receipts( + receipt_path: str, remaining: List[Dict[str, Any]] +) -> None: + """Drop reverted rows, removing the ledger entirely once it is empty. + + An empty ledger and an absent one mean the same thing to `adopt` — nothing + from this night is adopted — and removing the file lets the night be adopted + again cleanly. + """ + if remaining: + _write_atomic( + receipt_path, + json.dumps(remaining, ensure_ascii=False, indent=2), + create_parents=False, + ) + return + if os.path.lexists(receipt_path): + _unlink_fsync(receipt_path) + + +def _revert_one( + staging_dir: str, key: str, live: str, row: Dict[str, Any] +) -> RevertedTarget: + """Reverse one receipt row, or refuse if the live file moved on since.""" + before = row.get("sha256_before") + after = row.get("sha256_after") + backup_path = row.get("backup_path") + if not _valid_sha256_pin(after) or not isinstance(backup_path, str): + raise StagingError(f"adoption receipt for {key} has invalid pins") + if before != "" and not _valid_sha256_pin(before): + raise StagingError(f"adoption receipt for {key} has an invalid baseline pin") + + current, _mode, _file_id = _file_snapshot(live) + current_sha = _bytes_sha256(current) + if current_sha == before: + # An interrupted revert already published this path; finish the ledger. + return RevertedTarget(key, live, before if before else "", before == "", True) + if current_sha != after: + raise StagingError( + f"{live} no longer matches what adoption wrote for {key}; it was " + "edited or replaced since, and reverting would discard that work" + ) + + if before == "": + # Adoption created this file. The state being returned to is "no such + # file", so the proposal is removed rather than left in place. Any + # directories adoption created are left alone: the receipt does not + # record which ones it made, and removing a directory whose ownership + # was not durably recorded is exactly the fail-closed case adoption's + # own recovery path refuses. + if os.path.lexists(live): + if _is_link_or_junction(live) or not os.path.isfile(live): + raise StagingError(f"live path for {key} is no longer a regular file: {live}") + _unlink_fsync(live) + return RevertedTarget(key, live, "", True, False) + + if not backup_path: + raise StagingError( + f"adoption receipt for {key} records a replaced file but no backup" + ) + if _immutable_backup_sha256(backup_path, staging_dir) != before: + raise StagingError( + f"immutable backup for {key} is missing or does not match its pin: " + f"{backup_path}" + ) + original, _backup_mode, _backup_id = _file_snapshot(backup_path) + if original is None: + raise StagingError(f"immutable backup for {key} disappeared: {backup_path}") + try: + text = original.decode("utf-8") + except UnicodeDecodeError as exc: + raise StagingError(f"immutable backup for {key} is not valid UTF-8") from exc + _write_atomic(live, text, create_parents=False) + republished, _mode, _file_id = _file_snapshot(live) + if _bytes_sha256(republished) != before: + raise StagingError(f"live target for {key} changed while being restored") + # The backup has been consumed: its content is the live file again. Removing + # it returns the night to its pre-adopt shape, so it can be adopted afresh + # (adoption refuses to run when an immutable backup is already present). + if os.path.lexists(backup_path): + _unlink_fsync(backup_path) + return RevertedTarget(key, live, before, False, False) + + +def _revert_receipts( + staging_dir: str, + *, + filename: str, + key_field: str, + path_field: str, + selection: Optional[Sequence[str]], +) -> List[RevertedTarget]: + staging_dir = _canonical_staging_dir(staging_dir) + _recover_before_manifest(staging_dir) + receipt_path = os.path.join(staging_dir, filename) + rows = _receipt_rows(staging_dir, filename) + if not rows: + return [] + + wanted = None if selection is None else {str(name) for name in selection} + chosen: List[tuple[str, str, Dict[str, Any]]] = [] + live_paths: List[str] = [] + for row in rows: + key = str(row.get(key_field) or "") + if not key or (wanted is not None and key not in wanted): + continue + live = _safe_live_path(row.get(path_field)) + if not live: + raise StagingError(f"adoption receipt for {key} has an unsafe live path") + chosen.append((key, live, row)) + live_paths.append(live) + if wanted is not None: + missing = sorted(wanted - {key for key, _live, _row in chosen}) + if missing: + raise StagingError( + "not adopted from this night: " + ", ".join(repr(m) for m in missing) + ) + if not chosen: + return [] + + with _adoption_locks(staging_dir, live_paths): + # Re-read under the lock: another process may have reverted or adopted + # between our scan and the lock being taken. + current_rows = _receipt_rows(staging_dir, filename) + current_keys = {str(row.get(key_field) or "") for row in current_rows} + if current_keys != {str(row.get(key_field) or "") for row in rows}: + raise StagingError("adoption receipt changed while revert was locking") + results = [ + _revert_one(staging_dir, key, live, row) for key, live, row in chosen + ] + reverted = {result.key for result in results} + remaining = [ + row for row in current_rows + if str(row.get(key_field) or "") not in reverted + ] + _publish_remaining_receipts(receipt_path, remaining) + return results + + +def revert(staging_dir: str) -> List[RevertedTarget]: + """Undo `adopt` for this night's managed skill/memory pair.""" + return _revert_receipts( + staging_dir, + filename=_LEGACY_RECEIPT_FILE, + key_field="target", + path_field="live_path", + selection=None, + ) + + +def revert_skills( + staging_dir: str, skill_names: Optional[Sequence[str]] = None +) -> List[RevertedTarget]: + """Undo `adopt_skills` for a reviewed subset, or for every adopted skill.""" + return _revert_receipts( + staging_dir, + filename=_SKILLS_RECEIPT_FILE, + key_field="skill_name", + path_field="live_skill_path", + selection=skill_names, + ) + + +def adopted_skill_names(staging_dir: str) -> List[str]: + """Names of per-skill proposals adopted from this night and not yet reverted.""" + return [ + str(row.get("skill_name") or "") + for row in _receipt_rows(_canonical_staging_dir(staging_dir), _SKILLS_RECEIPT_FILE) + if row.get("skill_name") + ] + + +def has_adopted_legacy(staging_dir: str) -> bool: + """Whether the managed skill/memory pair is adopted and not yet reverted.""" + return bool( + _receipt_rows(_canonical_staging_dir(staging_dir), _LEGACY_RECEIPT_FILE) + ) + + +def latest_adopted_staging(project: str) -> Optional[str]: + """The newest staging directory holding an adoption that can still be undone. + + Deliberately not `latest_staging`: the newest proposal on disk may never + have been adopted, and reverting that would restore a backup for a change + the live files never received. + """ + root = staging_root(project) + if not os.path.isdir(root): + return None + subs = sorted( + (os.path.join(root, d) for d in os.listdir(root)), + key=lambda p: os.path.getmtime(p), + reverse=True, + ) + for path in subs: + if not os.path.isdir(path): + continue + try: + if has_adopted_legacy(path) or adopted_skill_names(path): + return path + except (OSError, StagingError): + continue + return None diff --git a/tests/test_sleep_revert.py b/tests/test_sleep_revert.py new file mode 100644 index 00000000..c7f3a293 --- /dev/null +++ b/tests/test_sleep_revert.py @@ -0,0 +1,410 @@ +"""Tests for undoing an adoption: `skillopt-sleep revert`. + +Revert is defined against the adoption receipts written by #212's durable +adoption transaction (`adopted_legacy.json` / `adopted_skills.json`), so these +tests drive real `adopt`/`adopt_skills` calls rather than hand-built ledgers. + +Pure-stdlib (unittest), hermetic (tmpdir only), no API key, no network. +Run: python -m pytest tests/test_sleep_revert.py +""" +from __future__ import annotations + +import io +import json +import os +import tempfile +import unittest +from contextlib import redirect_stdout + +from skillopt_sleep.__main__ import main +from skillopt_sleep.staging import ( + SkillProposal, + StagingError, + adopt, + adopt_skills, + adopted_skill_names, + has_adopted_legacy, + latest_adopted_staging, + revert, + revert_skills, + write_staging, +) +from skillopt_sleep.types import SleepReport + +ORIGINAL = "# hand-written skill\n\nAlways cite sources.\n" +PROPOSED = "# regressed skill\n\nJust guess.\n" + + +def _report(): + return SleepReport(night=1, project="/repo/example", accepted=True, + gate_action="accept_new_best") + + +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) + + +def _read(path): + with open(path, encoding="utf-8") as f: + return f.read() + + +def _stage_legacy(project, *, live_skill, skill=PROPOSED, + memory=None, live_memory=""): + return write_staging( + project, + report=_report(), + proposed_skill=skill, + proposed_memory=memory, + live_skill_path=live_skill, + live_memory_path=live_memory, + report_md="# report\n", + ) + + +def _stage_skills(project, proposals): + return write_staging( + project, + report=_report(), + proposed_skill=None, + proposed_memory=None, + live_skill_path="", + live_memory_path="", + report_md="# report\n", + skill_proposals=proposals, + ) + + +class TestRevertLegacy(unittest.TestCase): + def test_restores_the_pre_adopt_document(self): + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "skills", "learned", "SKILL.md") + _write(live, ORIGINAL) + staging = _stage_legacy(proj, live_skill=live) + + adopt(staging) + self.assertEqual(_read(live), PROPOSED) + + results = revert(staging) + self.assertEqual(_read(live), ORIGINAL) + self.assertEqual([r.key for r in results], ["skill"]) + self.assertFalse(results[0].removed) + self.assertFalse(results[0].already_reverted) + + def test_removes_a_file_adoption_created(self): + # The pre-adopt state was "no such file", so leaving the proposal in + # place would undo nothing. + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "skills", "learned", "SKILL.md") + os.makedirs(os.path.dirname(live)) + staging = _stage_legacy(proj, live_skill=live) + + adopt(staging) + self.assertTrue(os.path.exists(live)) + + results = revert(staging) + self.assertFalse(os.path.exists(live)) + self.assertTrue(results[0].removed) + + def test_restores_skill_and_memory_together(self): + with tempfile.TemporaryDirectory() as proj: + live_skill = os.path.join(proj, "SKILL.md") + live_memory = os.path.join(proj, "CLAUDE.md") + _write(live_skill, ORIGINAL) + _write(live_memory, "# memory\n") + staging = _stage_legacy(proj, live_skill=live_skill, + memory="# new memory\n", live_memory=live_memory) + + adopt(staging) + results = revert(staging) + + self.assertEqual(_read(live_skill), ORIGINAL) + self.assertEqual(_read(live_memory), "# memory\n") + self.assertEqual({r.key for r in results}, {"skill", "memory"}) + + def test_clears_the_receipt_so_the_night_can_be_adopted_again(self): + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + staging = _stage_legacy(proj, live_skill=live) + + adopt(staging) + self.assertTrue(has_adopted_legacy(staging)) + revert(staging) + self.assertFalse(has_adopted_legacy(staging)) + + # Adoption refuses to run while an immutable backup is present, so + # this also proves revert consumed the backup it restored from. + adopt(staging) + self.assertEqual(_read(live), PROPOSED) + revert(staging) + self.assertEqual(_read(live), ORIGINAL) + + def test_revert_converges_when_run_twice(self): + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + staging = _stage_legacy(proj, live_skill=live) + + adopt(staging) + revert(staging) + self.assertEqual(revert(staging), []) + self.assertEqual(_read(live), ORIGINAL) + + +class TestRevertRefusesWhenItCannot(unittest.TestCase): + def test_never_adopted_is_a_no_op(self): + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + staging = _stage_legacy(proj, live_skill=live) + self.assertEqual(revert(staging), []) + self.assertEqual(_read(live), ORIGINAL) + + def test_refuses_when_the_live_file_was_edited_after_adoption(self): + # Restoring the backup would silently discard the user's later edit. + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + staging = _stage_legacy(proj, live_skill=live) + adopt(staging) + _write(live, PROPOSED + "\nplus my own edit\n") + + with self.assertRaises(StagingError) as ctx: + revert(staging) + self.assertIn("edited or replaced since", str(ctx.exception)) + self.assertIn("plus my own edit", _read(live)) + + def test_refuses_when_the_backup_is_missing(self): + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + staging = _stage_legacy(proj, live_skill=live) + adopt(staging) + os.remove(os.path.join(staging, "backup", "SKILL.md")) + + with self.assertRaises(StagingError) as ctx: + revert(staging) + self.assertIn("backup", str(ctx.exception)) + + def test_refuses_an_unknown_skill_name(self): + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "skills", "alpha", "SKILL.md") + _write(live, ORIGINAL) + staging = _stage_skills(proj, [SkillProposal("alpha", PROPOSED, live)]) + adopt_skills(staging, ["alpha"]) + + with self.assertRaises(StagingError) as ctx: + revert_skills(staging, ["beta"]) + self.assertIn("not adopted from this night", str(ctx.exception)) + + +class TestRevertSkills(unittest.TestCase): + def _two_skills(self, proj): + alpha = os.path.join(proj, "skills", "alpha", "SKILL.md") + beta = os.path.join(proj, "skills", "beta", "SKILL.md") + _write(alpha, ORIGINAL) + _write(beta, ORIGINAL) + staging = _stage_skills(proj, [ + SkillProposal("alpha", PROPOSED, alpha), + SkillProposal("beta", PROPOSED, beta), + ]) + return staging, alpha, beta + + def test_reverts_a_reviewed_subset_and_leaves_the_rest(self): + with tempfile.TemporaryDirectory() as proj: + staging, alpha, beta = self._two_skills(proj) + adopt_skills(staging, ["alpha", "beta"]) + + revert_skills(staging, ["alpha"]) + + self.assertEqual(_read(alpha), ORIGINAL) + self.assertEqual(_read(beta), PROPOSED) + self.assertEqual(adopted_skill_names(staging), ["beta"]) + + def test_reverts_every_adopted_skill(self): + with tempfile.TemporaryDirectory() as proj: + staging, alpha, beta = self._two_skills(proj) + adopt_skills(staging, ["alpha", "beta"]) + + revert_skills(staging) + + self.assertEqual(_read(alpha), ORIGINAL) + self.assertEqual(_read(beta), ORIGINAL) + self.assertEqual(adopted_skill_names(staging), []) + + def test_per_skill_and_legacy_ledgers_are_independent(self): + with tempfile.TemporaryDirectory() as proj: + managed = os.path.join(proj, "SKILL.md") + skill = os.path.join(proj, "skills", "alpha", "SKILL.md") + _write(managed, ORIGINAL) + _write(skill, ORIGINAL) + staging = write_staging( + proj, + report=_report(), + proposed_skill=PROPOSED, + proposed_memory=None, + live_skill_path=managed, + live_memory_path="", + report_md="# report\n", + skill_proposals=[SkillProposal("alpha", PROPOSED, skill)], + ) + adopt(staging) + adopt_skills(staging, ["alpha"]) + + revert(staging) + + self.assertEqual(_read(managed), ORIGINAL) + self.assertEqual(_read(skill), PROPOSED) + self.assertFalse(has_adopted_legacy(staging)) + self.assertEqual(adopted_skill_names(staging), ["alpha"]) + + +class TestLatestAdoptedStaging(unittest.TestCase): + def test_skips_nights_that_were_staged_but_never_adopted(self): + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + older = _stage_legacy(proj, live_skill=live) + adopt(older) + _stage_legacy(proj, live_skill=live, skill="# another\n") + + self.assertEqual(latest_adopted_staging(proj), older) + + def test_none_once_everything_is_reverted(self): + with tempfile.TemporaryDirectory() as proj: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + staging = _stage_legacy(proj, live_skill=live) + adopt(staging) + revert(staging) + + self.assertIsNone(latest_adopted_staging(proj)) + + def test_none_without_a_staging_root(self): + with tempfile.TemporaryDirectory() as proj: + self.assertIsNone(latest_adopted_staging(proj)) + + +class TestRevertCli(unittest.TestCase): + def _run(self, argv): + buf = io.StringIO() + with redirect_stdout(buf): + code = main(argv) + return code, buf.getvalue() + + def _common(self, proj, home): + return ["--project", proj, "--claude-home", os.path.join(home, ".claude")] + + def test_reverts_and_reports(self): + with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + adopt(_stage_legacy(proj, live_skill=live)) + + code, out = self._run(["revert", *self._common(proj, home)]) + self.assertEqual(code, 0) + self.assertIn("reverted", out) + self.assertIn("restored", out) + self.assertEqual(_read(live), ORIGINAL) + + def test_nothing_adopted_exits_nonzero(self): + with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + _stage_legacy(proj, live_skill=live) + + code, out = self._run(["revert", *self._common(proj, home)]) + self.assertEqual(code, 1) + self.assertIn("nothing to revert", out) + + def test_rejects_more_than_one_selection_mode(self): + with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + adopt(_stage_legacy(proj, live_skill=live)) + + code, out = self._run([ + "revert", "--legacy", "--all-skills", *self._common(proj, home), + ]) + self.assertEqual(code, 2) + self.assertIn("exactly one of", out) + + def test_requires_a_selection_when_both_kinds_are_adopted(self): + with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home: + managed = os.path.join(proj, "SKILL.md") + skill = os.path.join(proj, "skills", "alpha", "SKILL.md") + _write(managed, ORIGINAL) + _write(skill, ORIGINAL) + staging = write_staging( + proj, + report=_report(), + proposed_skill=PROPOSED, + proposed_memory=None, + live_skill_path=managed, + live_memory_path="", + report_md="# report\n", + skill_proposals=[SkillProposal("alpha", PROPOSED, skill)], + ) + adopt(staging) + adopt_skills(staging, ["alpha"]) + + code, out = self._run(["revert", *self._common(proj, home)]) + self.assertEqual(code, 2) + self.assertIn("--skill NAME", out) + self.assertEqual(_read(managed), PROPOSED) + + def test_skill_selection_round_trip(self): + with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home: + skill = os.path.join(proj, "skills", "alpha", "SKILL.md") + _write(skill, ORIGINAL) + staging = _stage_skills(proj, [SkillProposal("alpha", PROPOSED, skill)]) + adopt_skills(staging, ["alpha"]) + + code, _out = self._run([ + "revert", "--skill", "alpha", *self._common(proj, home), + ]) + self.assertEqual(code, 0) + self.assertEqual(_read(skill), ORIGINAL) + + def test_json_output_reports_what_it_did(self): + with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + adopt(_stage_legacy(proj, live_skill=live)) + + code, out = self._run(["revert", "--json", *self._common(proj, home)]) + self.assertEqual(code, 0) + payload = json.loads(out) + self.assertTrue(payload["ok"]) + self.assertEqual(payload["mode"], "legacy") + self.assertEqual(len(payload["reverted"]), 1) + self.assertFalse(payload["reverted"][0]["removed"]) + + def test_refusal_is_reported_not_swallowed(self): + with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + adopt(_stage_legacy(proj, live_skill=live)) + _write(live, "edited by hand after adopting\n") + + code, out = self._run(["revert", *self._common(proj, home)]) + self.assertEqual(code, 2) + self.assertIn("revert refused", out) + self.assertEqual(_read(live), "edited by hand after adopting\n") + + def test_status_names_the_revertable_night(self): + with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home: + live = os.path.join(proj, "SKILL.md") + _write(live, ORIGINAL) + adopt(_stage_legacy(proj, live_skill=live)) + + code, out = self._run(["status", "--json", *self._common(proj, home)]) + self.assertEqual(code, 0) + self.assertTrue(json.loads(out)["revertable_staging"]) + + +if __name__ == "__main__": + unittest.main()