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