From 5b7f47f579923b0a95dbcade71d949ab7758a51a Mon Sep 17 00:00:00 2001 From: "Bogdan (Dan) Baciu" Date: Mon, 10 Aug 2026 02:32:34 +0400 Subject: [PATCH] feat(sleep): adopt reviewed skill subsets safely --- docs/sleep/README.md | 4 + docs/sleep/multi-skill-staging.md | 90 ++++++++++ skillopt_sleep/staging.py | 140 ++++++++++++++- tests/test_sleep_adopt_skill_subset.py | 239 +++++++++++++++++++++++++ 4 files changed, 472 insertions(+), 1 deletion(-) 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 5b29ebcf..7f5be874 100644 --- a/docs/sleep/README.md +++ b/docs/sleep/README.md @@ -299,6 +299,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 fb6f8863..d4e4d63a 100644 --- a/skillopt_sleep/staging.py +++ b/skillopt_sleep/staging.py @@ -7,14 +7,16 @@ """ from __future__ import annotations +import hashlib import json import os import re import shutil +import stat import tempfile import time from dataclasses import dataclass -from typing import Any, Dict, Iterable, List, Optional +from typing import Any, Dict, Iterable, List, Optional, Sequence from skillopt_sleep.types import SleepReport @@ -310,12 +312,17 @@ 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) + existing_mode = ( + stat.S_IMODE(os.stat(path).st_mode) if os.path.exists(path) else None + ) 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()) + if existing_mode is not None: + os.chmod(tmp, existing_mode) os.replace(tmp, path) except BaseException: if os.path.exists(tmp): @@ -495,6 +502,137 @@ 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 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) + if not isinstance(manifest, dict): + raise StagingError("staging manifest must be a JSON object") + if "skills" not in manifest: + return [] + rows = manifest["skills"] + if not isinstance(rows, list): + raise StagingError("staging manifest 'skills' must be a list") + if any(not isinstance(row, dict) for row in rows): + raise StagingError("every staging manifest 'skills' row must be an object") + return rows + + +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}" + ) + proposed_file = row.get("proposed_file") + expected_file = proposal_filename(name) + if proposed_file != expected_file: + raise StagingError( + f"unsafe staged proposal filename for {name!r}: {proposed_file!r}; " + f"expected {expected_file!r}" + ) + staged = os.path.join(staging_dir, expected_file) + 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 = hashlib.sha256(original).hexdigest() if original is not None else "" + _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..86247386 --- /dev/null +++ b/tests/test_sleep_adopt_skill_subset.py @@ -0,0 +1,239 @@ +"""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 stat +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), []) + + def test_malformed_skills_manifest_shape_is_refused(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + manifest_path = os.path.join(night.staging, "manifest.json") + for malformed in ({"not": "a list"}, [{"skill_name": "alpha"}, "bad"]): + with open(manifest_path, encoding="utf-8") as f: + manifest = json.load(f) + manifest["skills"] = malformed + with open(manifest_path, "w", encoding="utf-8") as f: + json.dump(manifest, f) + with self.assertRaises(StagingError, msg=repr(malformed)): + staged_skills(night.staging) + + +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_manifest_proposal_filename_cannot_escape_staging(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + outside = os.path.join(tmp, "outside.md") + _write(outside, "# not a staged proposal\n") + manifest_path = os.path.join(night.staging, "manifest.json") + with open(manifest_path, encoding="utf-8") as f: + manifest = json.load(f) + manifest["skills"][0]["proposed_file"] = os.path.relpath( + outside, night.staging + ) + with open(manifest_path, "w", encoding="utf-8") as f: + json.dump(manifest, f) + with self.assertRaises(StagingError): + adopt_skills(night.staging, ["alpha"]) + self.assertEqual(_read(night.alpha_live), "# alpha v1\n") + + def test_adoption_preserves_existing_live_file_mode(self): + with tempfile.TemporaryDirectory() as tmp: + night = TwoSkillNight(tmp) + os.chmod(night.alpha_live, 0o640) + adopt_skills(night.staging, ["alpha"]) + self.assertEqual(stat.S_IMODE(os.stat(night.alpha_live).st_mode), 0o640) + + 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()