From 58522d7fcadf2c7041b3b17cfb7af1752b26c3f7 Mon Sep 17 00:00:00 2001 From: vam Date: Tue, 8 Sep 2026 10:08:21 +0800 Subject: [PATCH 1/2] fix(sleep): keep train and val splits non-empty --- skillopt_sleep/mine.py | 47 +++++++++++++++++++++++++------ tests/test_split_hardening_2x3.py | 37 +++++++++++++++++++++++- 2 files changed, 75 insertions(+), 9 deletions(-) diff --git a/skillopt_sleep/mine.py b/skillopt_sleep/mine.py index f435519f5..069d35f4f 100644 --- a/skillopt_sleep/mine.py +++ b/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 diff --git a/tests/test_split_hardening_2x3.py b/tests/test_split_hardening_2x3.py index d32cc36db..f63992c56 100644 --- a/tests/test_split_hardening_2x3.py +++ b/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.""" From 3e134caf5fe75b02f6f19610ac995b095c6ad2f5 Mon Sep 17 00:00:00 2001 From: vam Date: Sat, 10 Oct 2026 17:29:39 +0800 Subject: [PATCH 2/2] fix: retain task exposure across nightly split reassignment --- docs/sleep/README.md | 8 +++ skillopt_sleep/cycle.py | 4 ++ skillopt_sleep/state.py | 66 +++++++++++++++++++++ tests/test_split_exposure.py | 109 +++++++++++++++++++++++++++++++++++ 4 files changed, 187 insertions(+) create mode 100644 tests/test_split_exposure.py diff --git a/docs/sleep/README.md b/docs/sleep/README.md index f47556eeb..4a18b0963 100644 --- a/docs/sleep/README.md +++ b/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 diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py index f1d7bf027..b140c2838 100644 --- a/skillopt_sleep/cycle.py +++ b/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, diff --git a/skillopt_sleep/state.py b/skillopt_sleep/state.py index 267adedd5..18afcb9ec 100644 --- a/skillopt_sleep/state.py +++ b/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: diff --git a/tests/test_split_exposure.py b/tests/test_split_exposure.py new file mode 100644 index 000000000..edb37b0d7 --- /dev/null +++ b/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)