fix(sleep): never certify a gate verdict measured on the training tasks

skillopt_sleep/consolidate.py

@@ -45,30 +45,48 @@ class ConsolidationResult:
     holdout_detail: List[dict] = field(default_factory=list)  # per val task: hard/soft/resp/why
     reflect_raw: str = ""        # the optimizer's last raw reply (empty => reflect produced nothing)
     call_error: str = ""         # backend's last call error (timeout/auth/empty)
+    # True when the gate's val slice was not disjoint from the tasks reflect
+    # saw, so its comparison cannot detect overfitting. A night in this state
+    # stages edits but never certifies them.
+    holdout_leaked: bool = False
 
 
-def _split(tasks: List[TaskRecord]) -> Tuple[List[TaskRecord], List[TaskRecord]]:
-    """Return (train_tasks, val_tasks).
+def _split(tasks: List[TaskRecord]) -> Tuple[List[TaskRecord], List[TaskRecord], bool]:
+    """Return ``(train_tasks, val_tasks, holdout_leaked)``.
 
     train drives reflect; val gates updates. test is held out entirely from
     consolidation and is scored by the caller. Accepts legacy split names
     (replay->train, holdout->val) for robustness.
+
+    ``holdout_leaked`` is True when val is not disjoint from train — i.e. the
+    gate would score the very tasks the edits were derived from. A single mined
+    task that carries a train/val (or legacy) split always lands here; a lone
+    ``test`` task instead yields empty train/val and no gate, so it is not
+    flagged as leaked. Such a non-disjoint comparison cannot detect overfitting,
+    so the caller must not treat it as validation.
     """
     def _norm(s: str) -> str:
         return {"replay": "train", "holdout": "val"}.get(s, s)
 
     train = [t for t in tasks if _norm(t.split) == "train"]
     val = [t for t in tasks if _norm(t.split) == "val"]
+    leaked = False
     # Be robust if a split is empty: fall back so a night still does something,
     # but never silently use test as train or val. An all-test batch therefore
     # returns empty train/val (caller scores test separately; gate is a no-op).
     if not val:
         # Prefer train as the gate reference; otherwise any non-test tasks.
         # Do not fall back to the full task list (that would leak held-out test).
         val = train or [t for t in tasks if _norm(t.split) != "test"]
+        leaked = bool(val)
     if not train:
         train = val
-    return train, val
+        leaked = leaked or bool(train)
+    if not leaked and train and val:
+        train_ids = {t.id for t in train}
+        if any(t.id in train_ids for t in val):
+            leaked = True
+    return train, val, leaked
 
 
 def _holdout_detail(pairs: List[Tuple[TaskRecord, ReplayResult]]) -> List[dict]:
@@ -116,7 +134,7 @@ def consolidate(
     """
     from skillopt_sleep import evidence as evlog
     ev = evlog.get(backend)
-    train_tasks, val_tasks = _split(tasks)
+    train_tasks, val_tasks, holdout_leaked = _split(tasks)
     gate_off = str(gate_mode).strip().lower() in {"off", "none", "false", "greedy"}
     holdout_detail: List[dict] = []
 
@@ -295,6 +313,14 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str:
         # `accepted` is False makes the headline contradict the outcome.
         if not accepted and action in {"accept", "accept_new_best"}:
             action = "reject"
+        # The gate cannot certify an improvement it measured on the same tasks
+        # the edits were derived from -- that comparison cannot detect
+        # overfitting, which is exactly how a reward hack scores 1.0. Abstain
+        # rather than report a verdict the evidence does not support. Edits are
+        # still staged, so a human can review them.
+        if accepted and holdout_leaked:
+            action = "reject_unverified"
+            accepted = False
         # A per-target trial can improve and tentatively apply an edit, while a
         # later fresh final replay regresses. The returned documents already
         # roll back in that case; keep the edit bookkeeping/report consistent
@@ -342,4 +368,5 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str:
         holdout_detail=holdout_detail,
         reflect_raw=getattr(backend, "last_reflect_raw", "") or "",
         call_error=getattr(backend, "last_call_error", "") or "",
+        holdout_leaked=holdout_leaked,
     )

skillopt_sleep/cycle.py

@@ -173,13 +173,31 @@ def _render_report_md(report: SleepReport, cfg: SleepConfig) -> str:
         f"- tokens used: {report.tokens_used}",
         "",
     ]
+    gate_on = str(cfg.get("gate_mode", "on")).strip().lower() not in {
+        "off", "none", "false", "greedy",
+    }
+    leaked_banner = report.holdout_leaked and gate_on
+    if leaked_banner:
+        lines[-1:] = [
+            "> **Not validated.** The gate's validation slice was not disjoint "
+            "from the tasks the optimizer saw (an overlapping set), so the "
+            "comparison above cannot detect overfitting. Mine more tasks so a "
+            "disjoint validation slice exists. Any edits below are unverified "
+            "suggestions, not rejections.",
+            "",
+        ]
     if report.edits:
         lines.append("## Accepted edits")
         for e in report.edits:
             lines.append(f"- [{e.target}/{e.op}] {e.content}  \n  _why: {e.rationale}_")
         lines.append("")
     if report.rejected_edits:
-        lines.append("## Rejected by gate (kept as negative feedback)")
+        # On a leaked-holdout night the gate abstained rather than rejecting, so
+        # these edits are unverified suggestions, not negative feedback.
+        if leaked_banner:
+            lines.append("## Unverified suggestions (not validated — review before adopting)")
+        else:
+            lines.append("## Rejected by gate (kept as negative feedback)")
         for e in report.rejected_edits:
             lines.append(f"- [{e.target}/{e.op}] {e.content}")
         lines.append("")
@@ -446,6 +464,7 @@ def run_sleep_cycle(
     report.candidate_score = result.candidate_score
     report.accepted = result.accepted
     report.gate_action = result.gate_action
+    report.holdout_leaked = getattr(result, "holdout_leaked", False)
     report.no_edits_reason = getattr(result, "no_edits_reason", "")
     report.edits = result.applied_edits
     report.rejected_edits = result.rejected_edits
@@ -494,6 +513,8 @@ def run_sleep_cycle(
                     "baseline_score": result.baseline_score,
                     "candidate_score": result.candidate_score,
                     "accepted": result.accepted,
+                    "gate_action": result.gate_action,
+                    "holdout_leaked": getattr(result, "holdout_leaked", False),
                     "n_applied_edits": len(result.applied_edits),
                     "n_rejected_edits": len(result.rejected_edits),
                     "n_unmatched_edits": len(result.unmatched_edits),

skillopt_sleep/types.py

@@ -143,6 +143,9 @@ class SleepReport:
     candidate_score: float = 0.0
     accepted: bool = False
     gate_action: str = ""
+    # True when the gate's validation slice was not disjoint from the tasks the
+    # optimizer saw, so its comparison could not detect overfitting.
+    holdout_leaked: bool = False
     no_edits_reason: str = ""
     edits: List[EditRecord] = field(default_factory=list)
     rejected_edits: List[EditRecord] = field(default_factory=list)

tests/test_consolidate_split.py

@@ -25,7 +25,7 @@ def test_all_test_batch_does_not_leak_into_train_or_val(self):
             TaskRecord(id="t1", project="p", intent="do Y", split="test"),
             TaskRecord(id="t2", project="p", intent="do Z", split="test"),
         ]
-        train, val = _split(tasks)
+        train, val, _leaked = _split(tasks)
         self.assertEqual(train, [])
         self.assertEqual(val, [])
 
@@ -41,7 +41,7 @@ def test_all_test_via_tasks_file_path_does_not_leak(self):
                 {"tasks": [t.to_dict() for t in tasks]},
             )
             loaded, _ = load_tasks_file(path)
-            train, val = _split(loaded)
+            train, val, _leaked = _split(loaded)
         self.assertEqual(_ids(train), [])
         self.assertEqual(_ids(val), [])
         self.assertEqual({t.split for t in loaded}, {"test"})
@@ -51,7 +51,7 @@ def test_train_only_falls_back_val_to_train(self):
             TaskRecord(id="a", project="p", intent="A", split="train"),
             TaskRecord(id="b", project="p", intent="B", split="train"),
         ]
-        train, val = _split(tasks)
+        train, val, _leaked = _split(tasks)
         self.assertEqual(_ids(train), ["a", "b"])
         self.assertEqual(_ids(val), ["a", "b"])
 
@@ -60,7 +60,7 @@ def test_train_plus_test_without_val_gates_on_train_not_test(self):
             TaskRecord(id="tr", project="p", intent="train", split="train"),
             TaskRecord(id="te", project="p", intent="test", split="test"),
         ]
-        train, val = _split(tasks)
+        train, val, _leaked = _split(tasks)
         self.assertEqual(_ids(train), ["tr"])
         self.assertEqual(_ids(val), ["tr"])
         self.assertNotIn("te", _ids(train) + _ids(val))
@@ -71,7 +71,7 @@ def test_explicit_train_val_test_keeps_partitions(self):
             TaskRecord(id="va", project="p", intent="val", split="val"),
             TaskRecord(id="te", project="p", intent="test", split="test"),
         ]
-        train, val = _split(tasks)
+        train, val, _leaked = _split(tasks)
         self.assertEqual(_ids(train), ["tr"])
         self.assertEqual(_ids(val), ["va"])
 
@@ -81,7 +81,7 @@ def test_legacy_holdout_name_maps_to_val(self):
             TaskRecord(id="va", project="p", intent="val", split="holdout"),
             TaskRecord(id="te", project="p", intent="test", split="test"),
         ]
-        train, val = _split(tasks)
+        train, val, _leaked = _split(tasks)
         self.assertEqual(_ids(train), ["tr"])
         self.assertEqual(_ids(val), ["va"])
 

tests/test_holdout_integrity.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+from skillopt_sleep.backend import Backend
+from skillopt_sleep.consolidate import _split, consolidate
+from skillopt_sleep.types import TaskRecord
+
+
+def _task(tid: str, split: str, rubric: str = "a good answer") -> TaskRecord:
+    return TaskRecord(
+        id=tid,
+        project="/p",
+        intent=f"intent {tid}",
+        reference_kind="rubric",
+        reference=rubric,
+        split=split,
+    )
+
+
+# --- holdout leak detection --------------------------------------------------
+
+
+def test_split_flags_leak_when_only_one_task() -> None:
+    train, val, leaked = _split([_task("a", "train")])
+    assert train and val
+    assert leaked is True
+
+
+def test_split_flags_leak_when_val_must_borrow_train() -> None:
+    _train, _val, leaked = _split([_task("a", "train"), _task("b", "train")])
+    assert leaked is True
+
+
+def test_split_flags_leak_when_train_must_borrow_val() -> None:
+    _train, _val, leaked = _split([_task("a", "val")])
+    assert leaked is True
+
+
+def test_split_flags_overlap_by_id() -> None:
+    shared = _task("a", "train")
+    dup = _task("a", "val")
+    _train, _val, leaked = _split([shared, dup])
+    assert leaked is True
+
+
+def test_split_is_clean_when_train_and_val_are_disjoint() -> None:
+    train, val, leaked = _split([_task("a", "train"), _task("b", "val")])
+    assert [t.id for t in train] == ["a"]
+    assert [t.id for t in val] == ["b"]
+    assert leaked is False
+
+
+def test_lone_test_task_is_never_used_as_train_or_val() -> None:
+    train, val, _leaked = _split([_task("t", "test")])
+    assert train == [] and val == []
+
+
+# --- end-to-end gate behaviour ----------------------------------------------
+
+
+class _ScriptedBackend(Backend):
+    """Backend that always proposes the one edit which helps.
+
+    Lets the gate be exercised without a live model: the candidate skill is the
+    one carrying MARKER, and the patched replay scores only that skill well.
+    """
+
+    name = "scripted"
+    MARKER = "ALWAYS REPORT WHAT WAS SEARCHED"
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.last_reflect_raw = ""
+        self.last_call_error = ""
+
+    def _call(self, prompt, *, max_tokens=1024):
+        return "response"
+
+    def reflect(self, failures, successes, skill, memory, *, edit_budget, evolve_skill, evolve_memory):
+        from skillopt_sleep.types import EditRecord
+
+        return [
+            EditRecord(
+                target="skill",
+                op="add",
+                content=self.MARKER,
+                anchor="",
+                rationale="report what was searched",
+            )
+        ]
+
+
+def test_gate_abstains_when_holdout_leaked(monkeypatch) -> None:
+    # One task means val == train, so an "improvement" cannot be validated.
+    from skillopt_sleep import consolidate as cons
+
+    def fake_replay_batch(backend, tasks, skill, memory, **kw):
+        from skillopt_sleep.types import ReplayResult
+
+        hard = 1.0 if _ScriptedBackend.MARKER in skill else 0.0
+        return [(t, ReplayResult(id=t.id, response="r", hard=hard, soft=hard)) for t in tasks]
+
+    monkeypatch.setattr(cons, "replay_batch", fake_replay_batch)
+    result = consolidate(
+        _ScriptedBackend(),
+        [_task("only", "train")],
+        skill="base skill",
+        memory="",
+        night=1,
+    )
+    assert result.holdout_leaked is True
+    assert result.accepted is False
+    assert result.gate_action == "reject_unverified"
+
+
+def test_gate_still_accepts_a_genuine_improvement(monkeypatch) -> None:
+    # The counterpart to the abstain test: a gate that rejects everything would
+    # be trivially safe and useless. With a disjoint val slice, a candidate that
+    # genuinely scores better on tasks the optimizer did NOT see must be
+    # accepted.
+    from skillopt_sleep import consolidate as cons
+
+    def fake_replay_batch(backend, tasks, skill, memory, **kw):
+        from skillopt_sleep.types import ReplayResult
+
+        hard = 1.0 if _ScriptedBackend.MARKER in skill else 0.0
+        return [(t, ReplayResult(id=t.id, response="r", hard=hard, soft=hard)) for t in tasks]
+
+    monkeypatch.setattr(cons, "replay_batch", fake_replay_batch)
+    result = consolidate(
+        _ScriptedBackend(),
+        [_task("tr", "train"), _task("va", "val")],
+        skill="base skill",
+        memory="",
+        night=1,
+    )
+    assert result.holdout_leaked is False
+    assert result.accepted is True
+    assert result.gate_action in {"accept", "accept_new_best"}
+    assert result.candidate_score > result.baseline_score
+    assert _ScriptedBackend.MARKER in result.new_skill
+
+
+# --- report banner -----------------------------------------------------------
+
+
+def test_not_validated_banner_shown_only_when_gate_is_on() -> None:
+    from skillopt_sleep.config import SleepConfig
+    from skillopt_sleep.cycle import _render_report_md
+    from skillopt_sleep.types import SleepReport
+
+    report = SleepReport(night=1, project="/p", holdout_leaked=True)
+
+    on = _render_report_md(report, SleepConfig())
+    assert "Not validated" in on
+
+    off_cfg = SleepConfig()
+    off_cfg.data["gate_mode"] = "off"
+    off = _render_report_md(report, off_cfg)
+    # In greedy mode the gate does no scoring, so the "gate scored the same
+    # tasks" banner would be misleading and must be suppressed.
+    assert "Not validated" not in off
+
+
+def test_leaked_edits_are_labeled_unverified_not_rejected() -> None:
+    from skillopt_sleep.config import SleepConfig
+    from skillopt_sleep.cycle import _render_report_md
+    from skillopt_sleep.types import EditRecord, SleepReport
+
+    edit = EditRecord(target="skill", op="add", content="a suggestion")
+
+    leaked = SleepReport(
+        night=1, project="/p", holdout_leaked=True, rejected_edits=[edit],
+    )
+    md = _render_report_md(leaked, SleepConfig())
+    # On a leaked-holdout night the gate abstained; the surfaced edits are
+    # unverified suggestions, not rejections/negative feedback.
+    assert "Unverified suggestions" in md
+    assert "negative feedback" not in md
+
+    genuine = SleepReport(
+        night=1, project="/p", holdout_leaked=False, rejected_edits=[edit],
+    )
+    md2 = _render_report_md(genuine, SleepConfig())
+    assert "negative feedback" in md2
+