fix(sleep): keep train and val splits non-empty

docs/sleep/README.md

@@ -339,6 +339,14 @@ correctness signal; the validation gate still governs what ships.
 | `recall_k` | `0` | Associative recall — pull the K most-similar past tasks (from a persisted archive) into tonight's dream. |
 | `dream_factor` | `0` | Add N lightweight synthetic variants of each task. |
 
+The nightly cycle persists task split exposure separately from the capped recall
+archive. Once a task has entered training, re-mining it or a recorded derivative
+cannot move it into validation or test; validation exposure likewise prevents
+later claims of pristine test coverage. This also applies to small-pool fallback
+assignments across growing or shrinking nightly pools. Restored assignments emit
+a warning when they reduce holdout coverage. If no disjoint validation pool
+remains, the existing gate reports leakage and does not certify the edits.
+
 ### Paired A/B evalkit
 
 Reports and PRs that claim "B beats A" should go through the shared evalkit

skillopt_sleep/cycle.py

@@ -812,6 +812,8 @@ def run_sleep_cycle(
             raise
         _progress(cfg, f"mine done: tasks={len(tasks)}")
 
+    state.protect_splits(tasks)
+
     if ev is not None:
         # Final task pool with split assignment: which tasks train the edits
         # vs. which held-out tasks gate them (works for seeded tasks too).
@@ -937,6 +939,8 @@ def run_sleep_cycle(
                 cfg, grouped, managed_name, skill
             )
             report.notes.extend(skip_notes)
+            for group in live_groups:
+                state.protect_splits(group.tasks)
             try:
                 consolidated_groups = consolidate_groups(
                     backend,

skillopt_sleep/mine.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import hashlib
+import logging
 import os
 import re
 from collections import Counter
@@ -24,6 +25,8 @@
 from skillopt_sleep.backend import CursorBackendError
 from skillopt_sleep.types import SessionDigest, TaskRecord
 
+_LOGGER = logging.getLogger("skillopt_sleep")
+
 
 def _tid(project: str, intent: str) -> str:
     h = hashlib.sha256((project + "::" + intent).encode("utf-8")).hexdigest()[:12]
@@ -286,7 +289,10 @@ def assign_splits(
         NEVER placed in val/test.
 
     A stable hash of the task id keeps the same real task in the same split across
-    nights (a fixed held-out gate, like SkillOpt's D_sel/D_test).
+    nights (a fixed held-out gate, like SkillOpt's D_sel/D_test). For a small
+    batch whose hash assignment cannot supply both train and val, the minimum
+    number of test tasks is reassigned and a warning records the reduced test
+    coverage.
 
     Back-compat: if ``test_fraction`` is 0 (default), this behaves like the old
     two-way replay/holdout split — real tasks divide into train + val, no test.
@@ -320,13 +326,18 @@ def _stable_key(task: TaskRecord) -> tuple[int, str]:
         bucket = int(hashlib.sha256((str(seed) + task.id).encode()).hexdigest(), 16)
         return bucket, task.id
 
+    promoted_from_test: List[Tuple[str, str]] = []
+
     def _promote_one(*, to: str, from_splits: set[str]) -> None:
-        """Promote one real task using hash order; never demote hash-assigned test."""
+        """Promote one real task using stable hash order."""
         candidates = [t for t in real if t.split in from_splits]
         if not candidates:
             return
         candidates.sort(key=_stable_key)
-        candidates[0].split = to
+        selected = candidates[0]
+        if selected.split == "test":
+            promoted_from_test.append((selected.id, to))
+        selected.split = to
 
     for t in real:
         bucket = _stable_key(t)[0] % 100
@@ -337,13 +348,33 @@ def _promote_one(*, to: str, from_splits: set[str]) -> None:
         else:
             t.split = "train"
 
-    # Guarantee val (the gate) is non-empty when we have >=2 real tasks.
-    # Only promote from train so hash-assigned test tasks stay untouched.
+    # Guarantee val (the gate) is non-empty when we have >=2 real tasks. Keep
+    # hash-assigned test tasks untouched unless the non-test pool cannot supply
+    # distinct train and val tasks.
     if len(real) >= 2 and not any(t.split == "val" for t in real):
-        _promote_one(to="val", from_splits={"train"})
-    # Guarantee a train pool exists when possible; never borrow from test.
+        real_train = [t for t in real if t.split == "train"]
+        train_count = sum(t.split == "train" for t in tasks)
+        if real_train and train_count >= 2:
+            _promote_one(to="val", from_splits={"train"})
+        else:
+            _promote_one(to="val", from_splits={"test"})
+    # Guarantee a train pool exists when possible. Prefer a spare val task; if
+    # the only val task is the gate, borrow from test so val stays non-empty.
     if not any(t.split == "train" for t in tasks) and len(real) >= 2:
-        _promote_one(to="train", from_splits={"val"})
+        val_count = sum(t.split == "val" for t in real)
+        if val_count >= 2:
+            _promote_one(to="train", from_splits={"val"})
+        else:
+            _promote_one(to="train", from_splits={"test"})
+
+    if promoted_from_test:
+        assignments = ", ".join(f"{task_id}->{split}" for task_id, split in promoted_from_test)
+        _LOGGER.warning(
+            "assign_splits reassigned %d test task(s) to keep train and val "
+            "non-empty; held-out test coverage was reduced (%s)",
+            len(promoted_from_test),
+            assignments,
+        )
     return tasks
 
 

skillopt_sleep/state.py

@@ -16,6 +16,8 @@
 import os
 from typing import Any, Dict, Optional
 
+from skillopt_sleep.types import TaskRecord
+
 
 def _now_iso(clock: Optional[float] = None) -> str:
     # caller passes a timestamp; we avoid importing time at module import
@@ -30,6 +32,8 @@ def _now_iso(clock: Optional[float] = None) -> str:
     "slow_memory": "",      # cross-night consolidated lessons (meta-skill analogue)
     "history": [],          # list of per-night summaries
     "task_archive": [],     # capped list of past mined tasks (for associative recall)
+    "split_exposure": {},   # uncapped project -> task id -> least-held-out split
+    "split_lineage": {},    # uncapped project -> child id -> parent ids
     "last_model_key": "",   # "backend::model" string used in the last successful night (F16)
     "last_model_key_format": 1,  # v1=config text; v2=resolved backend/model
 }
@@ -101,6 +105,68 @@ def add_to_archive(self, task_dicts: list, cap: int = 300) -> None:
         if len(arc) > cap:
             self.data["task_archive"] = arc[-cap:]
 
+    def protect_splits(self, tasks: list[TaskRecord]) -> None:
+        """Never promote previously exposed tasks or their lineage to holdout.
+
+        Keep provenance independently of the capped recall archive. Import old
+        archives on first use. Exposure is saved with the successful night's
+        state, preserving the cycle's existing abort/checkpoint semantics.
+        """
+        import logging
+
+        ranks = {"train": 0, "val": 1, "test": 2}
+        aliases = {"replay": "train", "holdout": "val"}
+        exposure = self.data.setdefault("split_exposure", {})
+        lineage = self.data.setdefault("split_lineage", {})
+        records = self.task_archive() + [t.to_dict() for t in tasks]
+        for row in records:
+            if row.get("derived_from"):
+                parents = lineage.setdefault(row.get("project", ""), {}).setdefault(row["id"], [])
+                if row["derived_from"] not in parents:
+                    parents.append(row["derived_from"])
+        # Propagate exposure through recorded parent/child links to a fixed
+        # point, including siblings that occur earlier in this night's pool.
+        changed = True
+        while changed:
+            changed = False
+            for row in records:
+                project = exposure.setdefault(row.get("project", ""), {})
+                ids = [row["id"]] + ([row["derived_from"]] if row.get("derived_from") else [])
+                split = aliases.get(row.get("split", "train"), row.get("split", "train"))
+                rank = min([ranks[split]] + [ranks[project.get(i, "test")] for i in ids])
+                if rank == 2:
+                    continue
+                for task_id in ids:
+                    if rank < ranks[project.get(task_id, "test")]:
+                        project[task_id] = ("train", "val")[rank]
+                        changed = True
+            for project_name, children in lineage.items():
+                project = exposure.setdefault(project_name, {})
+                for child, parents in children.items():
+                    ids = [child] + parents
+                    rank = min(ranks[project.get(i, "test")] for i in ids)
+                    for task_id in ids:
+                        if rank < ranks[project.get(task_id, "test")]:
+                            project[task_id] = ("train", "val")[rank]
+                            changed = True
+        for task in tasks:
+            previous = aliases.get(task.split, task.split)
+            protected = exposure.get(task.project, {}).get(task.id, previous)
+            if ranks[protected] < ranks[previous]:
+                logging.getLogger("skillopt_sleep").warning(
+                    "task %s retained as %s instead of %s due to persistent exposure; "
+                    "held-out coverage was reduced", task.id, protected, previous,
+                )
+                task.split = protected
+        # Check after restoring provenance: a test-hash task may now supply
+        # train. Only a genuinely missing train pool exposes val to reflect.
+        if tasks and not any(aliases.get(t.split, t.split) == "train" for t in tasks):
+            fallback = [t for t in tasks if aliases.get(t.split, t.split) == "val"]
+            if fallback:
+                for task in fallback:
+                    task.split = "train"
+                self.protect_splits(tasks)
+
     # ── model-swap tracking (F16) ─────────────────────────────────────────
     @property
     def last_model_key(self) -> str:

tests/test_split_exposure.py

@@ -0,0 +1,109 @@
+"""Cross-night split provenance through the real cycle and persisted archive."""
+from __future__ import annotations
+
+import hashlib
+import json
+from unittest import mock
+
+from skillopt_sleep.backend import build_backend
+from skillopt_sleep.config import load_config
+from skillopt_sleep.cycle import run_sleep_cycle
+from skillopt_sleep.state import SleepState
+from skillopt_sleep.types import TaskRecord
+
+
+def _task(task_id, project, **kwargs):
+    return TaskRecord(id=task_id, project=project, intent=f"task {task_id}",
+                      reference_kind="exact", reference=f"answer {task_id}", **kwargs)
+
+
+def _hash_id(split):
+    for index in range(1000):
+        task_id = f"normal-{index}"
+        bucket = int(hashlib.sha256(("42" + task_id).encode()).hexdigest(), 16) % 100
+        if ("val" if bucket < 10 else "test" if bucket < 90 else "train") == split:
+            return task_id
+    raise AssertionError("no matching hash bucket")
+
+
+def test_fallback_exposure_survives_growing_shrinking_and_archive_eviction(tmp_path):
+    project = str(tmp_path)
+    cfg = load_config(invoked_project=project, projects="invoked", backend="mock",
+                      state_dir=str(tmp_path / "state"), claude_home=str(tmp_path / ".claude"),
+                      val_fraction=0.10, test_fraction=0.80, seed=42)
+    backend = build_backend(backend="mock")
+    first_ids = [f"all-test-{i}" for i in range(1, 6)]
+    pools = [first_ids, first_ids + [_hash_id("train"), _hash_id("val")],
+             ["all-test-2", "all-test-4", _hash_id("val")]]
+    state_path = str(tmp_path / "state" / "state.json")
+    for night, ids in enumerate(pools):
+        # Only the external harvest/mining input is synthetic. Splitting,
+        # consolidation, evidence, archive writes and state reloads are real.
+        with mock.patch("skillopt_sleep.cycle.harvest_for_config", return_value=[]), \
+             mock.patch("skillopt_sleep.mine.heuristic_mine", return_value=[_task(i, project) for i in ids]), \
+             mock.patch("skillopt_sleep.cycle.build_backend", return_value=backend), \
+             mock.patch.object(backend, "reflect", wraps=backend.reflect) as reflect:
+            outcome = run_sleep_cycle(cfg)
+        with open(outcome.staging_dir + "/evidence.jsonl", encoding="utf-8") as handle:
+            rows = [json.loads(line) for line in handle]
+        ready = {r["task_id"]: r["split"] for r in rows if r["event"] == "task_ready"}
+        assert ready["all-test-2"] == "train", (night, ready)
+        assert any(t.id == "all-test-2" for call in reflect.call_args_list
+                   for pairs in call.args[:2] for t, _ in pairs)
+        held = [r for r in rows if r["event"] == "held_out_score"]
+        assert held[0]["n_test"] == sum(s == "test" for s in ready.values())
+        archived = SleepState.load(state_path).task_archive()
+        assert any(t["id"] == "all-test-2" and t["split"] == "train" for t in archived)
+
+    # Evict the raw task archive, then re-mine a derivative under a test hash.
+    state = SleepState.load(state_path)
+    state.add_to_archive([_task("unrelated", project).to_dict()], cap=1)
+    state.save()
+    derivative = _task(_hash_id("test"), project, derived_from="all-test-2", split="test")
+    outcome = run_sleep_cycle(cfg, seed_tasks=[derivative, _task("gate", project, split="val")])
+    assert derivative.split == "train"
+    assert not outcome.report.holdout_leaked
+
+
+def test_legacy_archive_exposure_is_loaded_before_scoring(tmp_path):
+    project = str(tmp_path)
+    cfg = load_config(invoked_project=project, projects="invoked", backend="mock",
+                      state_dir=str(tmp_path / "state"), claude_home=str(tmp_path / ".claude"))
+    state = SleepState.load(str(tmp_path / "state" / "state.json"))
+    state.add_to_archive([_task("old", project, split="replay").to_dict()])
+    state.save()
+    tasks = [_task("old", project, split="test"), _task("gate", project, split="val")]
+    run_sleep_cycle(cfg, seed_tasks=tasks)
+    assert tasks[0].split == "train"
+
+
+def test_lineage_survives_archive_eviction_and_is_project_scoped(tmp_path):
+    state_path = str(tmp_path / "state.json")
+    state = SleepState.load(state_path)
+    state.protect_splits([
+        _task("child", "/repo", split="test", derived_from="parent"),
+        _task("grandchild", "/repo", split="test", derived_from="child"),
+    ])
+    state.save()
+    state = SleepState.load(state_path)
+    state.protect_splits([_task("grandchild", "/repo", split="train")])
+    state.save()
+    state = SleepState.load(state_path)
+    tasks = [_task("parent", "/repo", split="test"),
+             _task("sibling", "/repo", split="val", derived_from="parent"),
+             _task("parent", "/other", split="test")]
+    state.protect_splits(tasks)
+    assert [t.split for t in tasks] == ["train", "train", "test"]
+
+
+def test_exposed_only_pool_cannot_be_certified_as_clean_holdout(tmp_path):
+    project = str(tmp_path)
+    cfg = load_config(invoked_project=project, projects="invoked", backend="mock",
+                      state_dir=str(tmp_path / "state"), claude_home=str(tmp_path / ".claude"))
+    run_sleep_cycle(cfg, seed_tasks=[_task("a", project), _task("b", project)])
+    outcome = run_sleep_cycle(cfg, seed_tasks=[_task("a", project, split="test"),
+                                               _task("b", project, split="val")])
+    assert outcome.report.holdout_leaked
+    assert not outcome.report.accepted
+    with open(outcome.staging_dir + "/evidence.jsonl", encoding="utf-8") as handle:
+        assert all(json.loads(line)["event"] != "held_out_score" for line in handle)

tests/test_split_hardening_2x3.py

@@ -20,8 +20,8 @@
 
 from skillopt_sleep.backend import build_backend
 from skillopt_sleep.config import load_config
-from skillopt_sleep.cycle import _resolve_split_fractions, run_sleep_cycle
 from skillopt_sleep.consolidate import consolidate
+from skillopt_sleep.cycle import _resolve_split_fractions, run_sleep_cycle
 from skillopt_sleep.dream import dream_consolidate, recall_similar
 from skillopt_sleep.mine import assign_splits
 from skillopt_sleep.state import SleepState
@@ -117,6 +117,41 @@ def test_hash_assigned_test_not_demoted_for_val_top_up(self):
             if t.id in test_ids:
                 self.assertEqual(t.split, "test")
 
+    def test_all_test_hash_assignment_keeps_train_and_val_nonempty(self):
+        tasks = [_task(f"all-test-{i}", f"all-test task {i}") for i in range(1, 6)]
+
+        with self.assertLogs("skillopt_sleep", level="WARNING") as logs:
+            out = assign_splits(
+                tasks,
+                val_fraction=0.10,
+                test_fraction=0.80,
+                seed=42,
+            )
+
+        self.assertEqual(sum(t.split == "train" for t in out), 1)
+        self.assertEqual(sum(t.split == "val" for t in out), 1)
+        self.assertEqual(sum(t.split == "test" for t in out), 3)
+        self.assertIn("reassigned 2 test task(s)", "\n".join(logs.output))
+
+    def test_val_only_hash_assignment_keeps_val_when_topping_up_train(self):
+        tasks = [
+            _task("val-only-26", "hashes to val"),
+            _task("all-test-1", "hashes to test"),
+            _task("all-test-2", "also hashes to test"),
+        ]
+
+        with self.assertLogs("skillopt_sleep", level="WARNING"):
+            out = assign_splits(
+                tasks,
+                val_fraction=0.10,
+                test_fraction=0.80,
+                seed=42,
+            )
+
+        self.assertEqual(sum(t.split == "train" for t in out), 1)
+        self.assertEqual(sum(t.split == "val" for t in out), 1)
+        self.assertEqual(sum(t.split == "test" for t in out), 1)
+
 
 class Pass1ApproachCFractionBoundaries(unittest.TestCase):
     """Pass 1 / approach C: reject invalid fraction knobs early."""