feat(sleep): stage one proposal and manifest row per skill

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,23 +399,32 @@ def write_staging(
     live_memory_path: str,
     report_md: str,
     out_dir: str = "",
+    skill_proposals: Sequence[SkillProposal] = (),
 ) -> str:
     """Write proposals + report into staging/<ts>/ 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,
         "has_skill": proposed_skill is not None,
         "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)

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