feat(sleep): adopt reviewed skill subsets safely

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).

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/<skill>/` 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.

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)