diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py
index 34335216..1c35fbcd 100644
--- a/skillopt_sleep/cycle.py
+++ b/skillopt_sleep/cycle.py
@@ -22,7 +22,12 @@
 from skillopt_sleep.dream import dream_consolidate
 from skillopt_sleep.harvest_sources import harvest_for_config
 from skillopt_sleep.memory import ensure_skill_scaffold
-from skillopt_sleep.mine import mine
+from skillopt_sleep.mine import group_tasks_by_skill_hint, mine
+from skillopt_sleep.multi_skill import (
+    SkillGroup,
+    consolidate_groups,
+    skill_group_reports,
+)
 from skillopt_sleep.staging import adopt as adopt_staging
 from skillopt_sleep.staging import redact_secrets
 from skillopt_sleep.staging import write_staging
@@ -211,6 +216,43 @@ def _render_report_md(report: SleepReport, cfg: SleepConfig) -> str:
             anchor = f"  \n  _anchor: `{e.anchor}`_" if e.anchor else ""
             lines.append(f"- [{e.target}/{e.op}] {e.content}{anchor}")
         lines.append("")
+    if report.skill_groups:
+        # The reviewer decides per skill, so the per-skill verdicts belong on
+        # the page they actually read. Without this the rows reach report.json
+        # only, and a human reviewing the night sees a single aggregate verdict
+        # that no individual skill necessarily earned.
+        lines.append("## Per-skill groups")
+        lines.append(
+            "_Each row is one skill's own evidence and its own gate decision. "
+            "A rejected group does not block its neighbours, and an accepted "
+            "one does not vouch for them._")
+        lines.append("")
+        lines.append("| Skill | Decision | Gate | Tasks | Held-out | Edits |")
+        lines.append("|---|---|---|---|---|---|")
+        for g in report.skill_groups:
+            name = g.skill_name or "_(no skill name)_"
+            if g.status == "consolidated":
+                decision = "accepted" if g.accepted else "rejected"
+                scores = f"{g.baseline_score:.3f} → {g.candidate_score:.3f}"
+                if g.gate_action == "reject_unverified":
+                    # This is the one score the gate explicitly refuses to
+                    # trust: it was measured on the same tasks the edits came
+                    # from, which is how a reward hack reaches 1.000. Printing
+                    # it bare reads as an improvement that was rejected for no
+                    # reason, so say why the number does not count.
+                    scores += " (unvalidated)"
+                edits = f"{g.n_applied_edits} applied / {g.n_rejected_edits} rejected"
+            else:
+                # skipped and failed groups never reached the gate; showing a
+                # 0.000 score for them would read as a measured result.
+                decision = g.status
+                scores = "—"
+                edits = "—"
+            reason = f" — {g.reason}" if g.reason else ""
+            lines.append(
+                f"| `{name}` | **{decision}**{reason} | {g.gate_action or '—'} "
+                f"| {g.n_tasks} | {scores} | {edits} |")
+        lines.append("")
     if report.notes:
         lines.append("## Notes")
         for n in report.notes:
@@ -469,6 +511,36 @@ def run_sleep_cycle(
     report.edits = result.applied_edits
     report.rejected_edits = result.rejected_edits
     report.unmatched_edits = result.unmatched_edits
+
+    # ── 4b. optional per-skill group reporting ───────────────────────────
+    # Off by default. When enabled, tonight's tasks are grouped by their skill
+    # hint and each group is consolidated independently so the report carries a
+    # row per skill instead of one aggregate verdict. This costs one extra
+    # consolidation per hinted group, which is why it is opt-in rather than
+    # automatic; a night whose evidence produces only the catch-all group adds
+    # no rows and no calls.
+    #
+    # Each group currently starts from the same managed document. Resolving a
+    # hinted group to its own live SKILL.md is the resolver's job and is not
+    # wired here yet, so a row describes what that group's evidence did to the
+    # managed skill, not to a separate file.
+    if cfg.get("multi_skill_report", False):
+        managed_name = cfg.get("managed_skill_name", "skillopt-sleep-learned")
+        grouped = group_tasks_by_skill_hint(tasks, managed_name)
+        if len(grouped) > 1:
+            _progress(cfg, f"multi-skill report: groups={len(grouped)}")
+            group_outcomes = consolidate_groups(
+                backend,
+                [SkillGroup(name, skill, rows) for name, rows in grouped.items()],
+                memory,
+                edit_budget=cfg.get("edit_budget", 4),
+                gate_metric=cfg.get("gate_metric", "mixed"),
+                gate_mixed_weight=cfg.get("gate_mixed_weight", 0.5),
+                gate_mode=cfg.get("gate_mode", "on"),
+                night=night,
+            )
+            report.skill_groups = skill_group_reports(group_outcomes)
+
     report.tokens_used = backend.tokens_used()
     report.ended_at = _now_iso(clock)
 
diff --git a/skillopt_sleep/staging.py b/skillopt_sleep/staging.py
index 12201ec9..41799a9b 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, Iterable, List, Optional
 
 from skillopt_sleep.types import SleepReport
 
@@ -234,6 +236,174 @@ 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 ""
+    # The name becomes a filename, so reject what Windows cannot store. Without
+    # this the write fails with an OSError from deep inside staging instead of
+    # a StagingError naming the offending skill.
+    if any(ch in candidate for ch in ':*?"<>|'):
+        return ""
+    if candidate[-1] in {".", " "}:
+        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 ""
+    raw = path.strip()
+    if raw.startswith("~"):
+        return ""
+    # Reject traversal on the RAW input, before normalising. Normalising first
+    # would silently resolve "/live/../../etc/SKILL.md" into "/etc/SKILL.md"
+    # and then accept it, because no ".." survives the collapse -- turning a
+    # traversal guard into a traversal helper.
+    if any(part in {os.curdir, os.pardir} for part in raw.replace("\\", "/").split("/")):
+        return ""
+    # Only then normalise, so a caller is not forced to hand over an already
+    # canonical string. The old form demanded input == normpath(input), which
+    # rejected benign duplicate separators and every forward-slash absolute
+    # path on Windows (normpath rewrites those to backslashes, so a safe path
+    # never matched itself).
+    candidate = os.path.normpath(raw)
+    if not os.path.isabs(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: Iterable[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 the skill name, the staged filename, or the live
+    path: a night must never stage two skills into one file or point a proposal
+    at the wrong one.
+
+    Staged filenames are compared case-insensitively. Skill names are
+    case-sensitive, so ``Research`` and ``research`` are two different skills on
+    a case-sensitive filesystem — but their proposal files land in one staging
+    directory, and on macOS and Windows that directory is case-insensitive, so
+    the second write silently replaces the first and the manifest then points a
+    surviving filename at another skill's content. Refusing the pair is the
+    conservative reading of the promise above.
+    """
+    rows: List[Dict[str, Any]] = []
+    seen_paths: Dict[str, str] = {}
+    seen_files: 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}")
+        proposed_file = proposal_filename(name)
+        # casefold, not lower: it folds Unicode pairs lower() leaves distinct,
+        # which is the comparison a case-insensitive filesystem actually makes.
+        file_key = proposed_file.casefold()
+        if file_key in seen_files:
+            raise StagingError(
+                f"skills {seen_files[file_key]!r} and {name!r} stage to the same "
+                f"file on a case-insensitive filesystem: {proposed_file}"
+            )
+        # Same reasoning for the live target: /x/A.md and /x/a.md are one file
+        # on macOS and Windows, so an exact-string check lets two skills
+        # overwrite each other's live document. casefold rather than
+        # os.path.normcase: normcase only folds case on Windows, so it is a
+        # no-op on the macOS box where the collision is just as real.
+        live_key = live.casefold()
+        if live_key in seen_paths:
+            raise StagingError(
+                f"skills {seen_paths[live_key]!r} and {name!r} target the same file: {live}"
+            )
+        seen_paths[live_key] = name
+        seen_files[file_key] = name
+        rows.append({
+            "skill_name": name,
+            "proposed_file": proposed_file,
+            "live_skill_path": live,
+        })
+    return rows
+
+
+def write_skill_proposals(
+    out_dir: str, proposals: Iterable[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.
+    """
+    # Materialise once. The signature accepts any Iterable, so a generator is
+    # legal input — and it would otherwise be drained by the validation pass,
+    # leaving the write loop with nothing to iterate and returning a full set
+    # of manifest rows for files that were never created.
+    proposals = list(proposals)
+    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 +449,23 @@ def write_staging(
     live_memory_path: str,
     report_md: str,
     out_dir: str = "",
+    skill_proposals: Iterable[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,
@@ -296,6 +473,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_engine.py b/tests/test_sleep_engine.py
index 9de9c8ea..feac2d3a 100644
--- a/tests/test_sleep_engine.py
+++ b/tests/test_sleep_engine.py
@@ -2377,3 +2377,139 @@ def test_group_tasks_by_skill_hint_hint_equal_to_managed_skill_is_one_group(self
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)
+
+
+class TestMultiSkillReportWiring(unittest.TestCase):
+    """The per-skill rows reach the emitted report (issue #120 follow-up)."""
+
+    def _hinted_tasks(self):
+        # Two skills' worth of evidence in one night. dataclasses.replace keeps
+        # the personas' real task shape rather than inventing a fixture.
+        from dataclasses import replace
+        research = assign_splits(researcher_persona(), holdout_fraction=0.34, seed=42)
+        programming = assign_splits(programmer_persona(), holdout_fraction=0.34, seed=1)
+        tagged = [replace(t, skill_hint="research-skill") for t in research]
+        tagged += [replace(t, id=f"prog-{t.id}", skill_hint="programming-skill")
+                   for t in programming]
+        return tagged
+
+    def test_multi_skill_report_is_off_by_default(self):
+        with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home:
+            cfg = load_config(
+                invoked_project=proj, projects="invoked", backend="mock",
+                claude_home=os.path.join(home, ".claude"),
+                managed_skill_name="skillopt-sleep-learned", auto_adopt=False,
+            )
+            outcome = run_sleep_cycle(cfg, seed_tasks=self._hinted_tasks())
+            # Opt-in: hinted evidence alone must not add rows or extra calls.
+            self.assertEqual(outcome.report.skill_groups, [])
+
+    def test_a_mixed_night_emits_one_independent_row_per_skill(self):
+        with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home:
+            cfg = load_config(
+                invoked_project=proj, projects="invoked", backend="mock",
+                claude_home=os.path.join(home, ".claude"),
+                managed_skill_name="skillopt-sleep-learned", auto_adopt=False,
+                multi_skill_report=True,
+            )
+            outcome = run_sleep_cycle(cfg, seed_tasks=self._hinted_tasks())
+            rows = outcome.report.skill_groups
+            names = [r.skill_name for r in rows]
+            self.assertIn("research-skill", names)
+            self.assertIn("programming-skill", names)
+
+            # Independence is the property under test: each row carries its own
+            # verdict and its own task count, not the night's aggregate.
+            for row in rows:
+                self.assertTrue(row.status)
+                self.assertGreater(row.n_tasks, 0)
+            self.assertEqual(len(rows), len(set(names)), "rows must not duplicate a skill")
+
+            # Independent verdicts, not one aggregate copied across rows: the
+            # two groups reach the gate on their own evidence and their own
+            # task counts.
+            self.assertTrue(all(r.status == "consolidated" for r in rows))
+            self.assertNotEqual(rows[0].n_tasks, rows[1].n_tasks)
+            self.assertNotEqual(rows[0].baseline_score, rows[1].baseline_score)
+
+    def test_a_mixed_night_reports_an_accepted_and_a_rejected_group(self):
+        # The case the maintainer asked for: one night, one group accepted and
+        # another rejected, each row carrying its own verdict. The rejected
+        # group is a genuine gate outcome rather than a contrived one — a group
+        # with a single task cannot validate, so the gate refuses to certify it.
+        from dataclasses import replace
+        tasks = self._hinted_tasks()
+        tasks.append(replace(tasks[0], id="thin-1", skill_hint="thin-skill"))
+        with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home:
+            cfg = load_config(
+                invoked_project=proj, projects="invoked", backend="mock",
+                claude_home=os.path.join(home, ".claude"),
+                managed_skill_name="skillopt-sleep-learned", auto_adopt=False,
+                multi_skill_report=True,
+            )
+            outcome = run_sleep_cycle(cfg, seed_tasks=tasks)
+            rows = {r.skill_name: r for r in outcome.report.skill_groups}
+            self.assertTrue(rows["research-skill"].accepted)
+            self.assertTrue(rows["programming-skill"].accepted)
+            self.assertFalse(rows["thin-skill"].accepted)
+            self.assertEqual(rows["thin-skill"].gate_action, "reject_unverified")
+            # The rejection is contained: it does not pull the others down.
+            self.assertEqual(rows["research-skill"].gate_action, "accept_new_best")
+
+    def test_report_md_shows_each_group_and_marks_an_uncertifiable_score(self):
+        # report.md is the page a human reads before /sleep adopt, so the rows
+        # have to reach it, not only report.json. And a reject_unverified score
+        # was measured on the tasks the edits came from — the comparison the
+        # gate refuses to certify — so it must not read as a plain improvement.
+        from dataclasses import replace
+        tasks = self._hinted_tasks()
+        tasks.append(replace(tasks[0], id="thin-1", skill_hint="thin-skill"))
+        with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home:
+            cfg = load_config(
+                invoked_project=proj, projects="invoked", backend="mock",
+                claude_home=os.path.join(home, ".claude"),
+                managed_skill_name="skillopt-sleep-learned", auto_adopt=False,
+                multi_skill_report=True,
+            )
+            outcome = run_sleep_cycle(cfg, seed_tasks=tasks)
+            with open(os.path.join(outcome.staging_dir, "report.md"),
+                      encoding="utf-8") as handle:
+                md = handle.read()
+            self.assertIn("## Per-skill groups", md)
+            self.assertIn("`research-skill`", md)
+            self.assertIn("`thin-skill`", md)
+            thin = [ln for ln in md.splitlines() if "`thin-skill`" in ln][0]
+            self.assertIn("rejected", thin)
+            self.assertIn("(unvalidated)", thin)
+            accepted = [ln for ln in md.splitlines() if "`research-skill`" in ln][0]
+            self.assertNotIn("(unvalidated)", accepted)
+
+    def test_report_md_has_no_group_section_when_the_feature_is_off(self):
+        with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home:
+            cfg = load_config(
+                invoked_project=proj, projects="invoked", backend="mock",
+                claude_home=os.path.join(home, ".claude"),
+                managed_skill_name="skillopt-sleep-learned", auto_adopt=False,
+            )
+            outcome = run_sleep_cycle(cfg, seed_tasks=self._hinted_tasks())
+            with open(os.path.join(outcome.staging_dir, "report.md"),
+                      encoding="utf-8") as handle:
+                self.assertNotIn("## Per-skill groups", handle.read())
+
+    def test_group_rows_survive_into_the_staged_report_json(self):
+        with tempfile.TemporaryDirectory() as proj, tempfile.TemporaryDirectory() as home:
+            cfg = load_config(
+                invoked_project=proj, projects="invoked", backend="mock",
+                claude_home=os.path.join(home, ".claude"),
+                managed_skill_name="skillopt-sleep-learned", auto_adopt=False,
+                multi_skill_report=True,
+            )
+            outcome = run_sleep_cycle(cfg, seed_tasks=self._hinted_tasks())
+            report_json = os.path.join(outcome.staging_dir, "report.json")
+            self.assertTrue(os.path.exists(report_json))
+            with open(report_json, encoding="utf-8") as handle:
+                payload = json.load(handle)
+            # Persisted, not merely present on the in-memory object.
+            self.assertTrue(payload.get("skill_groups"))
+            self.assertIn(payload["skill_groups"][0]["skill_name"],
+                          {"research-skill", "programming-skill"})
diff --git a/tests/test_sleep_staging_fanout.py b/tests/test_sleep_staging_fanout.py
new file mode 100644
index 00000000..dcc6dbb3
--- /dev/null
+++ b/tests/test_sleep_staging_fanout.py
@@ -0,0 +1,229 @@
+"""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_names_differing_only_by_case_are_refused(self):
+        # Skill names are case-sensitive, but every proposal lands in one
+        # staging directory — and on macOS and Windows that directory is
+        # case-insensitive. Before this guard, staging "Research" then
+        # "research" produced two manifest rows but one file on disk, named
+        # after the first skill and containing the second one's document.
+        with self.assertRaises(StagingError):
+            skill_proposal_rows([_proposal("Research"),
+                                 _proposal("research", live="/tmp/other/SKILL.md")])
+
+    def test_case_differing_proposals_never_lose_a_staged_file(self):
+        # Guards the guard: if the refusal above is ever relaxed, this asserts
+        # the actual filesystem outcome rather than the intent.
+        with tempfile.TemporaryDirectory() as out:
+            try:
+                rows = write_skill_proposals(
+                    out,
+                    [_proposal("Research"),
+                     _proposal("research", live="/tmp/other/SKILL.md")],
+                )
+            except StagingError:
+                return  # refused up front, which is the desired behaviour
+            staged = [f for f in os.listdir(out) if f.startswith("proposed_")]
+            self.assertEqual(len(staged), len(rows),
+                             "a manifest row exists whose staged file was overwritten")
+
+    def test_live_paths_differing_only_by_case_are_refused(self):
+        # os.path.normcase would not catch this: it only folds case on Windows,
+        # so it is a no-op on the macOS filesystem where /x/A.md and /x/a.md
+        # are nevertheless the same file.
+        with self.assertRaises(StagingError):
+            skill_proposal_rows([_proposal("alpha", live="/tmp/live/A.md"),
+                                 _proposal("beta", live="/tmp/live/a.md")])
+
+    def test_a_generator_of_proposals_still_writes_every_file(self):
+        # The annotation says Sequence but nothing enforces it. Validation used
+        # to drain a generator, leaving the write loop empty and returning a
+        # full set of manifest rows for files that were never created.
+        with tempfile.TemporaryDirectory() as out:
+            gen = (_proposal(n, live=f"/tmp/live/{n}/SKILL.md") for n in ("alpha", "beta"))
+            rows = write_skill_proposals(out, gen)
+            staged = [f for f in os.listdir(out) if f.startswith("proposed_")]
+            self.assertEqual(len(staged), len(rows))
+            self.assertEqual(len(rows), 2)
+
+    def test_names_windows_cannot_store_are_refused_cleanly(self):
+        # These reach the filesystem as a filename. Without an explicit guard
+        # they raise OSError from inside the write instead of a StagingError
+        # naming the offending skill.
+        for bad in ["a:b", "a*b", "a?b", 'a"b', "a<b", "a>b", "a|b", "trailing."]:
+            with self.assertRaises(StagingError, msg=bad):
+                skill_proposal_rows([_proposal(bad)])
+
+    def test_absolute_paths_needing_normalisation_are_accepted(self):
+        # Requiring the input to already equal normpath() rejected safe paths:
+        # duplicate separators everywhere, and every forward-slash absolute
+        # path on Windows. Normalising first keeps the traversal guard.
+        rows = skill_proposal_rows([_proposal("alpha", live="/tmp/live//alpha/SKILL.md")])
+        self.assertEqual(rows[0]["live_skill_path"], os.path.normpath("/tmp/live/alpha/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()