From fc7003da5ef25dd9534b2cba987e9a3d2c6d259c Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 28 Jul 2026 16:41:56 +0000 Subject: [PATCH 1/2] feat(sleep): stage one proposal and manifest row per skill Add SkillProposal, skill_proposal_rows, and write_skill_proposals: validate skill names, live target paths, and collisions before writing, then write each skill's proposal atomically. write_staging gains an optional skill_proposals fan-out and keeps the legacy single-proposal layout when it is unused. Refs #120 --- skillopt_sleep/staging.py | 131 +++++++++++++++++++++- tests/test_sleep_staging_fanout.py | 169 +++++++++++++++++++++++++++++ 2 files changed, 299 insertions(+), 1 deletion(-) create mode 100644 tests/test_sleep_staging_fanout.py diff --git a/skillopt_sleep/staging.py b/skillopt_sleep/staging.py index 12201ec9..bca36778 100644 --- a/skillopt_sleep/staging.py +++ b/skillopt_sleep/staging.py @@ -11,8 +11,10 @@ import os import re import shutil +import tempfile import time -from typing import Any, List, Optional +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, Sequence from skillopt_sleep.types import SleepReport @@ -234,6 +236,124 @@ def redact_secrets(value: Any) -> Any: return value +class StagingError(ValueError): + """A proposal could not be staged safely (bad name, bad target, collision).""" + + +@dataclass +class SkillProposal: + """One skill's proposed document plus the live file it would replace.""" + + skill_name: str + proposed_skill: str + live_skill_path: str + + +def _safe_skill_name(name: object) -> str: + """Return a skill name usable as a single path segment, else "".""" + if not isinstance(name, str): + return "" + candidate = name.strip() + if not candidate or candidate in {os.curdir, os.pardir}: + return "" + if candidate.startswith("~") or os.path.isabs(candidate): + return "" + if os.path.splitdrive(candidate)[0]: + return "" + separators = {"/", "\\", os.sep, os.altsep or os.sep} + if any(sep in candidate for sep in separators): + return "" + if any(ord(ch) < 32 or ord(ch) == 127 for ch in candidate): + return "" + return candidate + + +def _safe_live_path(path: object) -> str: + """Return an absolute, traversal-free ``*.md`` target path, else "".""" + if not isinstance(path, str) or not path.strip(): + return "" + candidate = path.strip() + if candidate.startswith("~") or not os.path.isabs(candidate): + return "" + if os.path.normpath(candidate) != candidate: + return "" + if not candidate.endswith(".md"): + return "" + return candidate + + +def proposal_filename(skill_name: str) -> str: + """Staged filename for one skill's proposal (unique per skill name).""" + return f"proposed_SKILL.{skill_name}.md" + + +def _write_atomic(path: str, text: str) -> None: + """Write ``text`` to ``path`` atomically, so review never sees half a file.""" + directory = os.path.dirname(path) or "." + os.makedirs(directory, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=directory, prefix=".tmp-", suffix=".md") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(text) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, path) + except BaseException: + if os.path.exists(tmp): + os.unlink(tmp) + raise + + +def skill_proposal_rows(proposals: Sequence[SkillProposal]) -> List[Dict[str, Any]]: + """Validate proposals and return their manifest rows, in input order. + + Raises :class:`StagingError` on an unusable skill name, an unsafe live target + path, or a collision on either the skill name or the live path: a night must + never stage two skills into one file or point a proposal at the wrong one. + """ + rows: List[Dict[str, Any]] = [] + seen_paths: Dict[str, str] = {} + for proposal in proposals: + name = _safe_skill_name(proposal.skill_name) + if not name: + raise StagingError(f"unsafe skill name for staging: {proposal.skill_name!r}") + live = _safe_live_path(proposal.live_skill_path) + if not live: + raise StagingError( + f"unsafe live skill path for {name!r}: {proposal.live_skill_path!r}" + ) + if any(row["skill_name"] == name for row in rows): + raise StagingError(f"duplicate skill name in staging fan-out: {name!r}") + if live in seen_paths: + raise StagingError( + f"skills {seen_paths[live]!r} and {name!r} target the same file: {live}" + ) + seen_paths[live] = name + rows.append({ + "skill_name": name, + "proposed_file": proposal_filename(name), + "live_skill_path": live, + }) + return rows + + +def write_skill_proposals( + out_dir: str, proposals: Sequence[SkillProposal] +) -> List[Dict[str, Any]]: + """Stage one uniquely named proposal file per skill; return manifest rows. + + Every proposal is validated before anything is written, so a rejected + fan-out leaves no partial files behind. + """ + rows = skill_proposal_rows(proposals) + if not rows: + return rows + os.makedirs(out_dir, exist_ok=True) + for row, proposal in zip(rows, proposals): + _write_atomic(os.path.join(out_dir, row["proposed_file"]), proposal.proposed_skill) + return rows + + def _ts_dir() -> str: return time.strftime("%Y%m%d-%H%M%S", time.localtime()) @@ -279,16 +399,23 @@ def write_staging( live_memory_path: str, report_md: str, out_dir: str = "", + skill_proposals: Sequence[SkillProposal] = (), ) -> str: """Write proposals + report into staging// and return that path. ``out_dir`` lets the cycle pre-create the night's staging folder at cycle START, so incremental artifacts (evidence.jsonl) accumulate in the same place the report lands. + + ``skill_proposals`` stages one extra uniquely named file and manifest row per + skill for a multi-skill night. Left empty, the staging layout and manifest + are exactly the legacy single-proposal ones. """ out = out_dir or os.path.join(staging_root(project), _ts_dir()) os.makedirs(out, exist_ok=True) + skill_rows = write_skill_proposals(out, skill_proposals) + manifest = { "live_skill_path": live_skill_path, "live_memory_path": live_memory_path, @@ -296,6 +423,8 @@ def write_staging( "has_memory": proposed_memory is not None, "accepted": report.accepted, } + if skill_rows: + manifest["skills"] = skill_rows if proposed_skill is not None: with open(os.path.join(out, "proposed_SKILL.md"), "w", encoding="utf-8") as f: f.write(proposed_skill) diff --git a/tests/test_sleep_staging_fanout.py b/tests/test_sleep_staging_fanout.py new file mode 100644 index 00000000..88b9527e --- /dev/null +++ b/tests/test_sleep_staging_fanout.py @@ -0,0 +1,169 @@ +"""Tests for per-skill staging fan-out (issue #120). + +Pure-stdlib (unittest), hermetic (tmpdir only), no API key, no network. +Run: python -m pytest tests/test_sleep_staging_fanout.py +""" +from __future__ import annotations + +import json +import os +import tempfile +import unittest + +from skillopt_sleep.staging import ( + SkillProposal, + StagingError, + proposal_filename, + skill_proposal_rows, + write_skill_proposals, + write_staging, +) +from skillopt_sleep.types import SleepReport + + +def _proposal(name="example-skill", body="# example\n", live=None, root="/tmp/live"): + if live is None: + live = os.path.join(root, name, "SKILL.md") + return SkillProposal(name, body, live) + + +def _report(): + return SleepReport(night=1, project="/repo/example", accepted=True, + gate_action="accept_new_best") + + +class TestSkillProposalRows(unittest.TestCase): + def test_one_row_per_skill_in_order(self): + rows = skill_proposal_rows([_proposal("alpha"), _proposal("beta")]) + self.assertEqual([r["skill_name"] for r in rows], ["alpha", "beta"]) + self.assertEqual([r["proposed_file"] for r in rows], + ["proposed_SKILL.alpha.md", "proposed_SKILL.beta.md"]) + self.assertEqual(rows[0]["live_skill_path"], "/tmp/live/alpha/SKILL.md") + + def test_filenames_are_unique_per_skill(self): + self.assertNotEqual(proposal_filename("alpha"), proposal_filename("beta")) + + def test_duplicate_skill_name_is_refused(self): + with self.assertRaises(StagingError): + skill_proposal_rows([_proposal("alpha"), + _proposal("alpha", live="/tmp/other/SKILL.md")]) + + def test_two_skills_targeting_one_file_are_refused(self): + shared = "/tmp/live/shared/SKILL.md" + with self.assertRaises(StagingError): + skill_proposal_rows([_proposal("alpha", live=shared), + _proposal("beta", live=shared)]) + + def test_unsafe_skill_names_are_refused(self): + for bad in ["", " ", ".", "..", "../escape", "a/b", "a\\b", "/abs", + "~home", "bad\nname"]: + with self.assertRaises(StagingError, msg=bad): + skill_proposal_rows([_proposal(bad)]) + + def test_unsafe_live_paths_are_refused(self): + for bad in ["", "relative/SKILL.md", "~/skills/a/SKILL.md", + "/tmp/live/../../etc/SKILL.md", "/tmp/live/a/SKILL.txt"]: + with self.assertRaises(StagingError, msg=bad): + skill_proposal_rows([_proposal("alpha", live=bad)]) + + +class TestWriteSkillProposals(unittest.TestCase): + def test_writes_one_file_per_skill(self): + with tempfile.TemporaryDirectory() as tmp: + rows = write_skill_proposals(tmp, [ + _proposal("alpha", "# alpha\n"), + _proposal("beta", "# beta\n"), + ]) + self.assertEqual(sorted(os.listdir(tmp)), + ["proposed_SKILL.alpha.md", "proposed_SKILL.beta.md"]) + with open(os.path.join(tmp, rows[0]["proposed_file"]), encoding="utf-8") as f: + self.assertEqual(f.read(), "# alpha\n") + + def test_no_proposals_writes_nothing(self): + with tempfile.TemporaryDirectory() as tmp: + self.assertEqual(write_skill_proposals(tmp, []), []) + self.assertEqual(os.listdir(tmp), []) + + def test_rejected_fan_out_leaves_no_partial_files(self): + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(StagingError): + write_skill_proposals(tmp, [_proposal("alpha"), _proposal("../escape")]) + self.assertEqual(os.listdir(tmp), []) + + def test_writes_leave_no_temporary_files_behind(self): + with tempfile.TemporaryDirectory() as tmp: + write_skill_proposals(tmp, [_proposal("alpha")]) + self.assertEqual([n for n in os.listdir(tmp) if n.startswith(".tmp-")], []) + + def test_rewrite_replaces_content_atomically(self): + with tempfile.TemporaryDirectory() as tmp: + write_skill_proposals(tmp, [_proposal("alpha", "# first\n")]) + write_skill_proposals(tmp, [_proposal("alpha", "# second\n")]) + path = os.path.join(tmp, proposal_filename("alpha")) + with open(path, encoding="utf-8") as f: + self.assertEqual(f.read(), "# second\n") + self.assertEqual(sorted(os.listdir(tmp)), [proposal_filename("alpha")]) + + +class TestWriteStagingCompatibility(unittest.TestCase): + def _manifest(self, out): + with open(os.path.join(out, "manifest.json"), encoding="utf-8") as f: + return json.load(f) + + def test_legacy_layout_when_multi_skill_is_unused(self): + with tempfile.TemporaryDirectory() as tmp: + out = write_staging( + tmp, report=_report(), proposed_skill="# skill\n", + proposed_memory="# memory\n", + live_skill_path=os.path.join(tmp, "live", "SKILL.md"), + live_memory_path=os.path.join(tmp, "live", "CLAUDE.md"), + report_md="# report\n", + ) + self.assertEqual( + sorted(os.listdir(out)), + ["manifest.json", "proposed_CLAUDE.md", "proposed_SKILL.md", + "report.json", "report.md"], + ) + manifest = self._manifest(out) + self.assertNotIn("skills", manifest) + self.assertTrue(manifest["has_skill"]) + + def test_fan_out_adds_files_and_manifest_rows(self): + with tempfile.TemporaryDirectory() as tmp: + live_root = os.path.join(tmp, "live") + out = write_staging( + tmp, report=_report(), proposed_skill=None, proposed_memory=None, + live_skill_path=os.path.join(live_root, "SKILL.md"), + live_memory_path=os.path.join(live_root, "CLAUDE.md"), + report_md="# report\n", + skill_proposals=[ + _proposal("alpha", "# alpha\n", root=live_root), + _proposal("beta", "# beta\n", root=live_root), + ], + ) + self.assertEqual( + sorted(os.listdir(out)), + ["manifest.json", "proposed_SKILL.alpha.md", "proposed_SKILL.beta.md", + "report.json", "report.md"], + ) + rows = self._manifest(out)["skills"] + self.assertEqual([r["skill_name"] for r in rows], ["alpha", "beta"]) + self.assertEqual(rows[1]["live_skill_path"], + os.path.join(live_root, "beta", "SKILL.md")) + + def test_unsafe_fan_out_writes_no_manifest(self): + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(StagingError): + write_staging( + tmp, report=_report(), proposed_skill=None, proposed_memory=None, + live_skill_path=os.path.join(tmp, "live", "SKILL.md"), + live_memory_path=os.path.join(tmp, "live", "CLAUDE.md"), + report_md="# report\n", + skill_proposals=[_proposal("alpha", live="relative/SKILL.md")], + ) + for root, _dirs, files in os.walk(tmp): + self.assertNotIn("manifest.json", files, root) + + +if __name__ == "__main__": + unittest.main() From 3779939dfcedf083070f463a704dacb96c395ba4 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 28 Jul 2026 16:44:42 +0000 Subject: [PATCH 2/2] feat(sleep): adopt a reviewed subset of staged skills with rollback Add staged_skills and adopt_skills: validate the selection first, back up and atomically write each live skill, roll the whole selection back on failure, and return/persist sha256 before/after receipts. Legacy single-proposal adoption is unchanged, and multi-skill staging plus migration notes are documented. Refs #120 --- docs/sleep/README.md | 4 + docs/sleep/multi-skill-staging.md | 90 +++++++++++ skillopt_sleep/staging.py | 125 +++++++++++++++ tests/test_sleep_adopt_skill_subset.py | 201 +++++++++++++++++++++++++ 4 files changed, 420 insertions(+) create mode 100644 docs/sleep/multi-skill-staging.md create mode 100644 tests/test_sleep_adopt_skill_subset.py diff --git a/docs/sleep/README.md b/docs/sleep/README.md index 2b740ac9..18312535 100644 --- a/docs/sleep/README.md +++ b/docs/sleep/README.md @@ -198,6 +198,10 @@ gate keeps the worst case bounded; keep it **on** by default. ## Learn more +Staging a proposal for more than one skill, and adopting a reviewed subset of +them with backups and hash receipts, is documented in +[`docs/sleep/multi-skill-staging.md`](multi-skill-staging.md). + See the [SkillOpt documentation index](../index.md), the [CLI reference](../reference/cli.md), and the integration-specific READMEs under [`plugins/`](https://github.com/microsoft/SkillOpt/tree/main/plugins). diff --git a/docs/sleep/multi-skill-staging.md b/docs/sleep/multi-skill-staging.md new file mode 100644 index 00000000..2f29e2fb --- /dev/null +++ b/docs/sleep/multi-skill-staging.md @@ -0,0 +1,90 @@ +# Multi-skill staging and subset adoption + +A night can stage a proposal for more than one skill. Adoption stays explicit: +staging only ever writes into the staging directory, and `adopt_skills()` copies +a **reviewed subset** over the live files, with a backup and a hash receipt per +skill. + +Nothing here changes a single-managed-skill night. If a night stages no per-skill +proposals, the staging directory and `manifest.json` are exactly the legacy ones +and `skillopt-sleep adopt` keeps working unchanged. + +## Staging layout + +Legacy (single managed skill) — unchanged: + +```text +.skillopt-sleep/staging/20260728-013000/ +├── manifest.json # live_skill_path, live_memory_path, has_skill, has_memory, accepted +├── proposed_SKILL.md +├── proposed_CLAUDE.md +├── report.json +└── report.md +``` + +Multi-skill night — one extra file and one manifest row per skill: + +```text +.skillopt-sleep/staging/20260728-013000/ +├── manifest.json # …the legacy keys plus "skills": [ … ] +├── proposed_SKILL.alpha.md +├── proposed_SKILL.beta.md +├── report.json # report.skill_groups carries each skill's gate evidence +└── report.md +``` + +```json +{ + "live_skill_path": "/home/dev/.claude/skills/alpha/SKILL.md", + "has_skill": false, + "accepted": true, + "skills": [ + { + "skill_name": "alpha", + "proposed_file": "proposed_SKILL.alpha.md", + "live_skill_path": "/home/dev/.claude/skills/alpha/SKILL.md" + }, + { + "skill_name": "beta", + "proposed_file": "proposed_SKILL.beta.md", + "live_skill_path": "/home/dev/.claude/skills/beta/SKILL.md" + } + ] +} +``` + +A skill name must be a single safe path segment and a live path must be an +absolute, traversal-free `*.md` file; two skills may not share a name or a target +file. A refused fan-out writes no `manifest.json`, so the folder is not adoptable. + +## Adopting a reviewed subset + +```python +from skillopt_sleep.staging import adopt_skills, latest_staging, staged_skills + +staging = latest_staging("/path/to/project") +[row["skill_name"] for row in staged_skills(staging)] # ['alpha', 'beta'] + +receipts = adopt_skills(staging, ["alpha"]) # beta is left alone +receipts[0].sha256_before, receipts[0].sha256_after +``` + +- `skill_names=None` adopts every staged skill; `[]` adopts nothing. +- An unknown or repeated name, an unsafe manifest row, or a missing proposal file + raises `StagingError` **before** anything is written. +- Each live file is backed up to `backup/skills//` and written atomically. +- If any write fails, every file in the selection is restored (and files that did + not exist before are removed), so a partial adoption never survives. +- Receipts (`skill_name`, `live_skill_path`, `sha256_before`, `sha256_after`, + `backup_path`) are returned and written to `adopted_skills.json` in the staging + directory. An empty `sha256_before` means the skill had no live file yet. + +## Migrating + +- **Consumers of `manifest.json`**: treat `"skills"` as optional; when absent the + night is a legacy single-proposal one. +- **Consumers of `report.json`**: `skill_groups` is `[]` on a single-skill night, + and the flat `accepted` / `gate_action` / score fields keep their meaning. +- **Adoption tooling**: `adopt()` still adopts the legacy single proposal pair. + Use `adopt_skills()` for per-skill nights; the two are independent, and neither + runs implicitly. diff --git a/skillopt_sleep/staging.py b/skillopt_sleep/staging.py index bca36778..82318275 100644 --- a/skillopt_sleep/staging.py +++ b/skillopt_sleep/staging.py @@ -7,6 +7,7 @@ """ from __future__ import annotations +import hashlib import json import os import re @@ -440,6 +441,130 @@ def write_staging( return out +@dataclass +class AdoptedSkill: + """Receipt for one adopted skill: where it landed and what changed.""" + + skill_name: str + live_skill_path: str + sha256_before: str # "" when no live file existed yet + sha256_after: str + backup_path: str = "" # "" when there was nothing to back up + + +def _sha256_text(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _sha256_file(path: str) -> str: + """Hash a file's contents, or "" when it does not exist.""" + if not os.path.exists(path): + return "" + with open(path, "rb") as f: + return hashlib.sha256(f.read()).hexdigest() + + +def staged_skills(staging_dir: str) -> List[Dict[str, Any]]: + """Manifest rows for the per-skill proposals staged in ``staging_dir``.""" + with open(os.path.join(staging_dir, "manifest.json"), encoding="utf-8") as f: + manifest = json.load(f) + rows = manifest.get("skills") or [] + return [row for row in rows if isinstance(row, dict)] + + +def _selected_rows( + rows: Sequence[Dict[str, Any]], skill_names: Optional[Sequence[str]] +) -> List[Dict[str, Any]]: + """Rows for the reviewed subset, in manifest order, or every row.""" + if skill_names is None: + return list(rows) + wanted = [str(n).strip() for n in skill_names] + if not wanted: + return [] + known = {str(row.get("skill_name", "")) for row in rows} + unknown = [n for n in wanted if n not in known] + if unknown: + raise StagingError(f"no staged proposal for: {', '.join(sorted(unknown))}") + duplicates = {n for n in wanted if wanted.count(n) > 1} + if duplicates: + raise StagingError(f"skill selected twice: {', '.join(sorted(duplicates))}") + chosen = set(wanted) + return [row for row in rows if str(row.get("skill_name", "")) in chosen] + + +def adopt_skills( + staging_dir: str, skill_names: Optional[Sequence[str]] = None +) -> List[AdoptedSkill]: + """Adopt an explicitly reviewed subset of staged per-skill proposals. + + ``skill_names`` selects which staged skills to adopt; ``None`` means every + staged skill. Nothing is adopted implicitly and skills outside the selection + are never touched. + + Every selected proposal is validated first, each live file is backed up, and + the writes are rolled back as a set if any one of them fails, so a partial + adoption never survives. Returns a before/after sha256 receipt per skill and + also writes them to ``adopted_skills.json`` in the staging directory. + """ + rows = _selected_rows(staged_skills(staging_dir), skill_names) + if not rows: + return [] + + plan: List[tuple] = [] + for row in rows: + name = _safe_skill_name(row.get("skill_name")) + if not name: + raise StagingError(f"unsafe staged skill name: {row.get('skill_name')!r}") + live = _safe_live_path(row.get("live_skill_path")) + if not live: + raise StagingError( + f"unsafe live skill path for {name!r}: {row.get('live_skill_path')!r}" + ) + staged = os.path.join(staging_dir, str(row.get("proposed_file") or "")) + if not os.path.isfile(staged): + raise StagingError(f"staged proposal missing for {name!r}: {staged}") + plan.append((name, live, staged)) + + backup_dir = os.path.join(staging_dir, "backup", "skills") + receipts: List[AdoptedSkill] = [] + done: List[tuple] = [] # (live, original_bytes or None) for rollback + try: + for name, live, staged in plan: + with open(staged, encoding="utf-8") as f: + proposed = f.read() + original = None + backup_path = "" + if os.path.exists(live): + with open(live, "rb") as f: + original = f.read() + skill_backup = os.path.join(backup_dir, name) + os.makedirs(skill_backup, exist_ok=True) + backup_path = os.path.join(skill_backup, os.path.basename(live)) + shutil.copy2(live, backup_path) + before = _sha256_file(live) + _write_atomic(live, proposed) + done.append((live, original)) + receipts.append(AdoptedSkill( + skill_name=name, live_skill_path=live, sha256_before=before, + sha256_after=_sha256_text(proposed), backup_path=backup_path, + )) + except BaseException: + for live, original in reversed(done): + if original is None: + if os.path.exists(live): + os.unlink(live) + else: + with open(live, "wb") as f: + f.write(original) + raise + + _write_atomic( + os.path.join(staging_dir, "adopted_skills.json"), + json.dumps([r.__dict__ for r in receipts], ensure_ascii=False, indent=2), + ) + return receipts + + def _backup(path: str, backup_dir: str) -> None: if os.path.exists(path): os.makedirs(backup_dir, exist_ok=True) diff --git a/tests/test_sleep_adopt_skill_subset.py b/tests/test_sleep_adopt_skill_subset.py new file mode 100644 index 00000000..cc1320d2 --- /dev/null +++ b/tests/test_sleep_adopt_skill_subset.py @@ -0,0 +1,201 @@ +"""Tests for explicit multi-skill subset adoption (issue #120). + +Pure-stdlib (unittest), hermetic (tmpdir only), no API key, no network. +Run: python -m pytest tests/test_sleep_adopt_skill_subset.py +""" +from __future__ import annotations + +import hashlib +import json +import os +import tempfile +import unittest + +from skillopt_sleep.staging import ( + SkillProposal, + StagingError, + adopt_skills, + staged_skills, + write_staging, +) +from skillopt_sleep.types import SleepReport + + +def _sha(text): + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _read(path): + with open(path, encoding="utf-8") as f: + return f.read() + + +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 TwoSkillNight: + """End-to-end fixture: a staged night with two per-skill proposals.""" + + def __init__(self, tmp): + self.tmp = tmp + self.live_root = os.path.join(tmp, "live") + self.alpha_live = os.path.join(self.live_root, "alpha", "SKILL.md") + self.beta_live = os.path.join(self.live_root, "beta", "SKILL.md") + _write(self.alpha_live, "# alpha v1\n") + _write(self.beta_live, "# beta v1\n") + self.staging = write_staging( + tmp, + report=SleepReport(night=1, project=tmp, accepted=True), + proposed_skill=None, proposed_memory=None, + live_skill_path=self.alpha_live, + live_memory_path=os.path.join(self.live_root, "CLAUDE.md"), + report_md="# report\n", + skill_proposals=[ + SkillProposal("alpha", "# alpha v2\n", self.alpha_live), + SkillProposal("beta", "# beta v2\n", self.beta_live), + ], + ) + + +class TestStagedSkills(unittest.TestCase): + def test_rows_are_readable_from_the_manifest(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + rows = staged_skills(night.staging) + self.assertEqual([r["skill_name"] for r in rows], ["alpha", "beta"]) + + def test_legacy_single_proposal_night_has_no_staged_skills(self): + with tempfile.TemporaryDirectory() as tmp: + out = write_staging( + tmp, report=SleepReport(night=1, project=tmp), proposed_skill="# s\n", + proposed_memory=None, + live_skill_path=os.path.join(tmp, "live", "SKILL.md"), + live_memory_path=os.path.join(tmp, "live", "CLAUDE.md"), + report_md="# report\n", + ) + self.assertEqual(staged_skills(out), []) + self.assertEqual(adopt_skills(out), []) + + +class TestAdoptSkillSubset(unittest.TestCase): + def test_adopting_one_skill_leaves_the_other_untouched(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + receipts = adopt_skills(night.staging, ["alpha"]) + self.assertEqual([r.skill_name for r in receipts], ["alpha"]) + self.assertEqual(_read(night.alpha_live), "# alpha v2\n") + self.assertEqual(_read(night.beta_live), "# beta v1\n") + + def test_receipts_carry_before_and_after_hashes(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + receipt = adopt_skills(night.staging, ["alpha"])[0] + self.assertEqual(receipt.sha256_before, _sha("# alpha v1\n")) + self.assertEqual(receipt.sha256_after, _sha("# alpha v2\n")) + self.assertEqual(receipt.live_skill_path, night.alpha_live) + self.assertEqual(_read(receipt.backup_path), "# alpha v1\n") + + def test_receipts_are_persisted_beside_the_report(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + adopt_skills(night.staging, ["beta"]) + with open(os.path.join(night.staging, "adopted_skills.json"), + encoding="utf-8") as f: + rows = json.load(f) + self.assertEqual([r["skill_name"] for r in rows], ["beta"]) + self.assertEqual(rows[0]["sha256_after"], _sha("# beta v2\n")) + + def test_selecting_no_skills_adopts_nothing(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + self.assertEqual(adopt_skills(night.staging, []), []) + self.assertEqual(_read(night.alpha_live), "# alpha v1\n") + self.assertEqual(_read(night.beta_live), "# beta v1\n") + self.assertFalse( + os.path.exists(os.path.join(night.staging, "adopted_skills.json"))) + + def test_selecting_every_skill_adopts_all_of_them(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + receipts = adopt_skills(night.staging) + self.assertEqual([r.skill_name for r in receipts], ["alpha", "beta"]) + self.assertEqual(_read(night.alpha_live), "# alpha v2\n") + self.assertEqual(_read(night.beta_live), "# beta v2\n") + + def test_a_new_live_file_reports_an_empty_before_hash(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + os.unlink(night.beta_live) + receipt = [r for r in adopt_skills(night.staging) if r.skill_name == "beta"][0] + self.assertEqual(receipt.sha256_before, "") + self.assertEqual(receipt.backup_path, "") + self.assertEqual(_read(night.beta_live), "# beta v2\n") + + def test_unknown_or_repeated_selection_is_refused_without_writing(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + for selection in (["gamma"], ["alpha", "gamma"], ["alpha", "alpha"]): + with self.assertRaises(StagingError, msg=str(selection)): + adopt_skills(night.staging, selection) + self.assertEqual(_read(night.alpha_live), "# alpha v1\n") + self.assertEqual(_read(night.beta_live), "# beta v1\n") + + def test_missing_staged_proposal_is_refused_without_writing(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + os.unlink(os.path.join(night.staging, "proposed_SKILL.beta.md")) + with self.assertRaises(StagingError): + adopt_skills(night.staging) + self.assertEqual(_read(night.alpha_live), "# alpha v1\n") + + def test_unsafe_manifest_row_is_refused_without_writing(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + manifest_path = os.path.join(night.staging, "manifest.json") + with open(manifest_path, encoding="utf-8") as f: + manifest = json.load(f) + manifest["skills"][1]["live_skill_path"] = "relative/SKILL.md" + with open(manifest_path, "w", encoding="utf-8") as f: + json.dump(manifest, f) + with self.assertRaises(StagingError): + adopt_skills(night.staging) + self.assertEqual(_read(night.alpha_live), "# alpha v1\n") + + def test_a_failed_write_rolls_the_whole_selection_back(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + # beta's live path becomes un-writable: its parent is now a file. + os.unlink(night.beta_live) + os.rmdir(os.path.dirname(night.beta_live)) + _write(os.path.dirname(night.beta_live), "not a directory\n") + with self.assertRaises(OSError): + adopt_skills(night.staging) + self.assertEqual(_read(night.alpha_live), "# alpha v1\n") + self.assertFalse( + os.path.exists(os.path.join(night.staging, "adopted_skills.json"))) + + def test_rollback_removes_files_that_did_not_exist_before(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + os.unlink(night.alpha_live) + os.unlink(night.beta_live) + os.rmdir(os.path.dirname(night.beta_live)) + _write(os.path.dirname(night.beta_live), "not a directory\n") + with self.assertRaises(OSError): + adopt_skills(night.staging) + self.assertFalse(os.path.exists(night.alpha_live)) + + def test_adoption_never_happens_without_an_explicit_call(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + self.assertEqual(_read(night.alpha_live), "# alpha v1\n") + self.assertEqual(_read(night.beta_live), "# beta v1\n") + self.assertTrue(os.path.exists( + os.path.join(night.staging, "proposed_SKILL.alpha.md"))) + + +if __name__ == "__main__": + unittest.main()