fix(sleep): honor val_fraction and test_fraction in the nightly cycle

skillopt_sleep/config.py

@@ -214,13 +214,20 @@ def _load_file(path: str) -> Dict[str, Any]:
 
 def load_config(**overrides: Any) -> SleepConfig:
     data = dict(DEFAULTS)
+    user_keys: set[str] = set()
     path = _user_config_path()
     if path:
         try:
-            data.update(_load_file(path) or {})
+            file_data = _load_file(path) or {}
+            user_keys.update(file_data.keys())
+            data.update(file_data)
         except Exception:
             pass
-    data.update({k: v for k, v in overrides.items() if v is not None})
+    for key, value in overrides.items():
+        if value is not None:
+            data[key] = value
+            user_keys.add(key)
     if data.get("projects") == "invoked" and not data.get("invoked_project"):
         data["invoked_project"] = os.getcwd()
+    data["_user_config_keys"] = sorted(user_keys)
     return SleepConfig(data=data)

skillopt_sleep/cycle.py

@@ -21,12 +21,13 @@
 
 from skillopt_sleep import evidence
 from skillopt_sleep.backend import Backend, CursorBackendError, build_backend
-from skillopt_sleep.config import SleepConfig, load_config
+from skillopt_sleep.config import DEFAULTS, SleepConfig, load_config
 from skillopt_sleep.dream import dream_consolidate
 from skillopt_sleep.evidence import EvidenceLog
 from skillopt_sleep.harvest_sources import harvest_for_config
 from skillopt_sleep.memory import ensure_skill_scaffold
 from skillopt_sleep.mine import group_tasks_by_skill_hint, mine
+from skillopt_sleep.replay import aggregate_scores, replay_batch
 from skillopt_sleep.multi_skill import (
     SKIPPED,
     GroupConsolidation,
@@ -53,6 +54,41 @@
 )
 
 
+def _resolve_split_fractions(cfg: SleepConfig) -> tuple[float, float]:
+    """Return ``(val_fraction, test_fraction)`` for tonight's split.
+
+    ``holdout_fraction`` is the documented legacy alias for ``val_fraction``.
+    ``load_config`` records which keys came from the user's file or explicit
+    overrides in ``_user_config_keys``; the alias applies only when the user
+    set ``holdout_fraction`` and did not set ``val_fraction``. An explicit
+    ``val_fraction`` (including ``0.0``) always wins over the alias.
+
+    Raises ``ValueError`` when fractions are out of range or their sum is >= 1.
+    """
+    user_keys = set(cfg.get("_user_config_keys") or ())
+
+    if "val_fraction" in user_keys:
+        val = float(cfg.get("val_fraction"))
+    else:
+        val = float(DEFAULTS["val_fraction"])
+        if "holdout_fraction" in user_keys:
+            val = float(cfg.get("holdout_fraction"))
+
+    if "test_fraction" in user_keys:
+        test = float(cfg.get("test_fraction"))
+    else:
+        test = float(DEFAULTS["test_fraction"])
+
+    for name, value in (("val_fraction", val), ("test_fraction", test)):
+        if not 0.0 <= value <= 1.0:
+            raise ValueError(f"{name} must be between 0 and 1 inclusive, got {value}")
+    if val + test >= 1.0:
+        raise ValueError(
+            f"val_fraction + test_fraction must be < 1 (got {val} + {test})"
+        )
+    return val, test
+
+
 # ── Model-swap detection (F16) ───────────────────────────────
 def _make_model_key(cfg: SleepConfig) -> str:
     """Stable string identifying the backend object(s) actually used.
@@ -658,7 +694,7 @@ def run_sleep_cycle(
             "backend", "model", "optimizer_backend", "optimizer_model",
             "target_backend", "target_model", "gate_mode", "gate_metric",
             "gate_mixed_weight", "gate_no_regression", "edit_budget",
-            "holdout_fraction",
+            "holdout_fraction", "val_fraction", "test_fraction",
             "dream_rollouts", "dream_factor", "recall_k",
             "max_tasks_per_night", "lookback_hours", "llm_mine",
             "evolve_skill", "evolve_memory")}
@@ -758,12 +794,14 @@ def run_sleep_cycle(
             f"mine start: max_tasks={max_tasks} candidate_limit={candidate_limit} "
             f"llm_mine={llm_miner is not None} target_filter={target_filter}",
         )
+        val_fraction, test_fraction = _resolve_split_fractions(cfg)
         try:
             tasks = mine(
                 digests,
                 max_tasks=max_tasks,
                 candidate_limit=candidate_limit,
-                holdout_fraction=cfg.get("holdout_fraction", 0.34),
+                val_fraction=val_fraction,
+                test_fraction=test_fraction,
                 seed=cfg.get("seed", 42),
                 llm_miner=llm_miner,
                 target_skill_text=raw_skill if target_filter else "",
@@ -854,6 +892,24 @@ def run_sleep_cycle(
     report.unmatched_edits = result.unmatched_edits
     report.gate_trials = redact_secrets(getattr(result, "gate_trials", []))
 
+    # ── held-out test measure (write-only; the gate never reads it) ──────
+    # consolidate() holds the test split out entirely and documents that the
+    # caller scores it; in the nightly, this is that caller. Scored on the
+    # night's FINAL documents (post-gate-decision), mirroring how the
+    # experiment harness reports its per-night test score. With the default
+    # test_fraction=0.0 no test task exists and this block never runs, so
+    # legacy nights are bit-for-bit unchanged and cost nothing extra.
+    test_tasks = [t for t in tasks if t.split == "test"]
+    if test_tasks and ev is not None:
+        final_skill = result.new_skill if result.accepted else skill
+        final_memory = result.new_memory if result.accepted else memory
+        test_pairs = replay_batch(backend, test_tasks, final_skill, final_memory)
+        test_hard, test_soft = aggregate_scores(test_pairs)
+        ev.log("test", "held_out_score", night=night,
+               n_test=len(test_tasks),
+               hard=round(test_hard, 4), soft=round(test_soft, 4),
+               accepted=result.accepted, gate_action=result.gate_action)
+
     # ── 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

skillopt_sleep/dream.py

@@ -59,19 +59,38 @@ def _tokens(text: str) -> set:
     return {w for w in re.findall(r"[a-z0-9]+", (text or "").lower()) if len(w) > 2}
 
 
-def recall_similar(new_tasks: List[TaskRecord], history: List[TaskRecord],
-                   k: int) -> List[TaskRecord]:
+def _normalize_split(value: str) -> str:
+    return {"replay": "train", "holdout": "val"}.get(value, value)
+
+
+def recall_similar(
+    new_tasks: List[TaskRecord],
+    history: List[TaskRecord],
+    k: int,
+    *,
+    exclude_ids: Optional[set[str]] = None,
+) -> List[TaskRecord]:
     """Return the ``k`` historical tasks most lexically similar to any of
     tonight's ``new_tasks`` (max Jaccard token overlap). Recalled tasks are
     returned as training material (split='train'); deterministic, stdlib-only.
+
+    Archived val/test tasks are never recalled, and ``exclude_ids`` blocks
+    tonight's held-out ids (and their ``derived_from`` sources) from re-entering
+    the training pool.
     """
     if not history or k <= 0 or not new_tasks:
         return []
+    blocked = set(exclude_ids or ())
+    for t in new_tasks:
+        blocked.add(t.id)
+        if t.derived_from:
+            blocked.add(t.derived_from)
     new_tok = [_tokens(t.intent) for t in new_tasks]
-    new_ids = {t.id for t in new_tasks}
     scored = []
     for h in history:
-        if h.id in new_ids:
+        if h.id in blocked:
+            continue
+        if _normalize_split(h.split) in ("val", "test"):
             continue
         ht = _tokens(h.intent)
         if not ht:
@@ -127,7 +146,15 @@ def dream_consolidate(
     train = [t for t in tasks if t.split == "train"]
     enlarged = list(tasks)
     if recall_k > 0 and history_tasks:
-        enlarged += recall_similar(train, history_tasks, recall_k)
+        held_out_ids = {
+            t.id for t in tasks if _normalize_split(t.split) in ("val", "test")
+        }
+        for t in tasks:
+            if t.derived_from:
+                held_out_ids.add(t.derived_from)
+        enlarged += recall_similar(
+            train, history_tasks, recall_k, exclude_ids=held_out_ids,
+        )
     if dream_factor > 0:
         seed = [t for t in enlarged if t.split == "train" and t.origin != "dream"]
         enlarged += dream_augment(seed, factor=dream_factor)

skillopt_sleep/mine.py

@@ -295,6 +295,17 @@ def assign_splits(
     if holdout_fraction is not None:
         val_fraction = holdout_fraction
 
+    for name, value in (("val_fraction", val_fraction), ("test_fraction", test_fraction)):
+        if not 0.0 <= value <= 1.0:
+            raise ValueError(
+                f"{name} must be between 0 and 1 inclusive, got {value}"
+            )
+    if val_fraction + test_fraction >= 1.0:
+        raise ValueError(
+            f"val_fraction + test_fraction must be < 1 "
+            f"(got {val_fraction} + {test_fraction})"
+        )
+
     dream = [t for t in tasks if t.origin == "dream"]
     real = [t for t in tasks if t.origin != "dream"]
 
@@ -304,28 +315,35 @@ def assign_splits(
 
     val_cut = int(round(val_fraction * 100))
     test_cut = val_cut + int(round(test_fraction * 100))
+
+    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."""
+        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
+
     for t in real:
-        bucket = int(hashlib.sha256((str(seed) + t.id).encode()).hexdigest(), 16) % 100
+        bucket = _stable_key(t)[0] % 100
         if bucket < val_cut:
             t.split = "val"
         elif bucket < test_cut:
             t.split = "test"
         else:
             t.split = "train"
 
-    # guarantee val (the gate) is non-empty when we have >=2 real tasks
-    real_splits = {t.split for t in real}
-    if len(real) >= 2 and "val" not in real_splits:
-        real[-1].split = "val"
-    # guarantee a train pool exists (dream or real) when possible
+    # Guarantee val (the gate) is non-empty when we have >=2 real tasks.
+    # Only promote from train so hash-assigned test tasks stay untouched.
+    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 any(t.split == "train" for t in tasks) and len(real) >= 2:
-        real[0].split = "train"
-    # if test was requested but ended up empty with >=3 real tasks, carve one
-    if test_fraction > 0 and len(real) >= 3 and not any(t.split == "test" for t in real):
-        for t in real:
-            if t.split == "train":
-                t.split = "test"
-                break
+        _promote_one(to="train", from_splits={"val"})
     return tasks
 
 
@@ -339,13 +357,20 @@ def mine(
     *,
     max_tasks: int = 40,
     candidate_limit: int = 0,
-    holdout_fraction: float = 0.34,
+    val_fraction: float = 0.34,
+    test_fraction: float = 0.0,
+    holdout_fraction: float | None = None,  # legacy alias for val_fraction
     seed: int = 42,
     llm_miner: Optional[Callable[[List[SessionDigest]], List[TaskRecord]]] = None,
     target_skill_text: str = "",
     target_skill_path: str = "",
 ) -> List[TaskRecord]:
-    """Top-level miner. Uses ``llm_miner`` if provided, else heuristic."""
+    """Top-level miner. Uses ``llm_miner`` if provided, else heuristic.
+
+    Split knobs mirror ``assign_splits``: ``val_fraction``/``test_fraction``
+    are the real controls; ``holdout_fraction`` remains the legacy alias and,
+    when passed, overrides ``val_fraction`` (same contract as assign_splits).
+    """
     candidate_limit = candidate_limit or max_tasks
     tasks: List[TaskRecord] = []
     if llm_miner is not None:
@@ -361,5 +386,11 @@ def mine(
     if target_skill_text or target_skill_path:
         tasks = filter_tasks_for_target(tasks, target_skill_text, target_skill_path)
     tasks = tasks[:max_tasks]
-    tasks = assign_splits(tasks, holdout_fraction=holdout_fraction, seed=seed)
+    tasks = assign_splits(
+        tasks,
+        val_fraction=val_fraction,
+        test_fraction=test_fraction,
+        holdout_fraction=holdout_fraction,
+        seed=seed,
+    )
     return tasks