fix(sleep): keep the gate non-empty when every task hashes into test

assign_splits promises "Guarantee val (the gate) is non-empty when we have
>=2 real tasks", but _promote_one only pulls from train (to top up val) or
from val (to top up train), never from test. When every real task's hash
bucket lands in [val_cut, test_cut) both train and val start empty, so both
guarantee calls have nothing to promote and silently no-op.

run_sleep_cycle then finishes with gate_action='reject' and edits=0: no
error, no warning, and holdout_leaked does not flag it either. For a small
nightly batch any test_fraction above roughly 0.5 makes this a matter of
when, not if.

Fall back to test for the promotion only when the preferred source is empty,
so the stability guarantee #235 settled is untouched in the normal case and
hash-assigned test tasks are only reassigned in the degenerate one. Topping
up train now prefers test over a single-task val, so filling train cannot
re-empty the gate it just filled. Borrowing spends a held-out task, so it is
logged the way consolidate.py already logs holdout_leaked.

Tests cover the degenerate split for 2..10 real tasks, the warning, dream
tasks in train not masking an empty gate, the normal path not borrowing, and
the single-real-task guard.

Author: linhongyu510

skillopt_sleep/mine.py

@@ -320,13 +320,18 @@ def _stable_key(task: TaskRecord) -> tuple[int, str]:
         bucket = int(hashlib.sha256((str(seed) + task.id).encode()).hexdigest(), 16)
         return bucket, task.id
 
-    def _promote_one(*, to: str, from_splits: set[str]) -> None:
-        """Promote one real task using hash order; never demote hash-assigned test."""
+    def _promote_one(*, to: str, from_splits: set[str]) -> bool:
+        """Promote one real task using hash order; never demote hash-assigned test.
+
+        Returns True when a task was promoted, so degenerate splits that had to
+        borrow from ``test`` can be reported rather than failing silently.
+        """
         candidates = [t for t in real if t.split in from_splits]
         if not candidates:
-            return
+            return False
         candidates.sort(key=_stable_key)
         candidates[0].split = to
+        return True
 
     for t in real:
         bucket = _stable_key(t)[0] % 100
@@ -338,12 +343,42 @@ def _promote_one(*, to: str, from_splits: set[str]) -> None:
             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.
+    # Prefer train so hash-assigned test tasks stay untouched. When every real
+    # task hashed into test there is nothing in train to promote, and the old
+    # code silently no-opped: the cycle then rejected with edits=0, no error,
+    # and holdout_leaked did not flag it either. Fall back to test in that
+    # degenerate case only, and log it since it spends a held-out task.
+    borrowed_from_test = False
     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.
+        if not _promote_one(to="val", from_splits={"train"}):
+            borrowed_from_test = _promote_one(to="val", from_splits={"test"})
+
+    # Guarantee a train pool exists when possible. Prefer val, but never empty
+    # the gate to do it: if val holds a single task, take from test first and
+    # only fall back to val when test is exhausted (the pre-existing behavior
+    # for splits that have no test slice at all).
     if not any(t.split == "train" for t in tasks) and len(real) >= 2:
-        _promote_one(to="train", from_splits={"val"})
+        spare_val = sum(1 for t in real if t.split == "val") > 1
+        if spare_val:
+            _promote_one(to="train", from_splits={"val"})
+        elif _promote_one(to="train", from_splits={"test"}):
+            borrowed_from_test = True
+        else:
+            _promote_one(to="train", from_splits={"val"})
+
+    if borrowed_from_test:
+        import logging
+
+        logging.getLogger("skillopt_sleep").warning(
+            "assign_splits: all %d real tasks hashed into test "
+            "(val_fraction=%.2f, test_fraction=%.2f, seed=%d); "
+            "borrowed from test so the gate has a val slice. "
+            "Lower test_fraction to stop spending held-out tasks.",
+            len(real),
+            val_fraction,
+            test_fraction,
+            seed,
+        )
     return tasks
 
 

tests/test_split_hardening_2x3.py

@@ -117,6 +117,83 @@ 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_val_guaranteed_when_every_real_task_hashes_into_test(self):
+        """Degenerate split: nothing in train/val to promote from.
+
+        With a high test_fraction a small nightly batch can hash entirely into
+        test. Both guarantees then had nothing to promote and silently no-opped,
+        so the cycle rejected with edits=0 and no error. The gate must still get
+        a val slice, borrowed from test as a last resort.
+        """
+        for n in range(2, 11):
+            with self.subTest(real_tasks=n):
+                tasks = assign_splits(
+                    [_task(f"t{i}", f"task {i}") for i in range(n)],
+                    val_fraction=0.10,
+                    test_fraction=0.80,
+                    seed=42,
+                )
+                splits = [t.split for t in tasks]
+                self.assertIn("val", splits, "gate must not be left empty")
+                self.assertIn("train", splits, "train pool must not be empty")
+
+    def test_degenerate_split_warns_that_test_was_spent(self):
+        """Borrowing from test is reported, not silent."""
+        with self.assertLogs("skillopt_sleep", level="WARNING") as captured:
+            assign_splits(
+                [_task(f"t{i}", f"task {i}") for i in range(5)],
+                val_fraction=0.10,
+                test_fraction=0.80,
+                seed=42,
+            )
+        self.assertTrue(
+            any("hashed into test" in line for line in captured.output),
+            f"expected a degenerate-split warning, got {captured.output}",
+        )
+
+    def test_dream_train_does_not_mask_an_empty_gate(self):
+        """Dream tasks fill train, but they may never stand in for val."""
+        real = [_task(f"t{i}", f"task {i}") for i in range(4)]
+        dream = [_task("d0", "dream variant", origin="dream")]
+        tasks = assign_splits(
+            real + dream,
+            val_fraction=0.10,
+            test_fraction=0.80,
+            seed=42,
+        )
+        val_rows = [t for t in tasks if t.split == "val"]
+        self.assertTrue(val_rows, "gate must be filled from real tasks")
+        for t in val_rows:
+            self.assertNotEqual(t.origin, "dream", "val must stay real-only")
+        for t in tasks:
+            if t.origin == "dream":
+                self.assertEqual(t.split, "train")
+
+    def test_normal_split_does_not_borrow_from_test(self):
+        """The healthy path keeps its hash-assigned test slice intact."""
+        with mock.patch("logging.getLogger") as get_logger:
+            tasks = assign_splits(
+                [_task(f"t{i}", f"task {i}") for i in range(12)],
+                val_fraction=0.34,
+                test_fraction=0.10,
+                seed=7,
+            )
+            get_logger.assert_not_called()
+        splits = [t.split for t in tasks]
+        self.assertIn("val", splits)
+        self.assertIn("train", splits)
+
+    def test_single_real_task_is_left_alone(self):
+        """The >=2 guard still applies; one task cannot fill val and train."""
+        tasks = assign_splits(
+            [_task("only", "task")],
+            val_fraction=0.10,
+            test_fraction=0.80,
+            seed=42,
+        )
+        self.assertEqual(len(tasks), 1)
+        self.assertIn(tasks[0].split, {"train", "val", "test"})
+
 
 class Pass1ApproachCFractionBoundaries(unittest.TestCase):
     """Pass 1 / approach C: reject invalid fraction knobs early."""