From f1b2c651781eaf1f80d307b727218b68a84d4f9c Mon Sep 17 00:00:00 2001 From: "Bogdan (Dan) Baciu" Date: Sun, 30 Aug 2026 02:13:19 -0600 Subject: [PATCH 1/2] feat(sleep): add adversarial candidate probes --- docs/sleep/README.md | 37 ++- docs/sleep/multi-skill-staging.md | 4 +- skillopt_sleep/adversarial.py | 224 ++++++++++++++++ skillopt_sleep/config.py | 3 + skillopt_sleep/consolidate.py | 121 ++++++++- skillopt_sleep/cycle.py | 85 +++++- skillopt_sleep/dream.py | 19 +- tests/test_adversarial_dream.py | 429 ++++++++++++++++++++++++++++++ tests/test_sleep_engine.py | 13 +- 9 files changed, 907 insertions(+), 28 deletions(-) create mode 100644 skillopt_sleep/adversarial.py create mode 100644 tests/test_adversarial_dream.py diff --git a/docs/sleep/README.md b/docs/sleep/README.md index f47556ee..529a3e65 100644 --- a/docs/sleep/README.md +++ b/docs/sleep/README.md @@ -308,7 +308,8 @@ This runs one additional consolidation per group (including a catch-all group wh hinted and unhinted evidence are mixed), so it multiplies backend calls and token use; configured dream rollouts and synthetic variants multiply the per-group work too. Each group inherits the configured edit budget, gate mode/metric, -`gate_no_regression`, `dream_rollouts`, `dream_factor`, `recall_k`, and +`gate_no_regression`, `dream_rollouts`, `dream_factor`, `dream_adversarial`, +`dream_adversarial_blocking`, `dream_adversarial_margin`, `recall_k`, and `evolve_skill`. Recalled archive tasks are restricted to that same skill hint; shared memory is read-only in fan-out runs. Setting `evolve_skill` to `false` therefore disables per-skill proposals as well as the managed skill proposal. @@ -327,17 +328,43 @@ Resolution searches existing project-native `.agents/skills`, `.claude/skills`, home and plugin-cache roots. Add repeatable `--skill-root PATH` values when an integration stores skills elsewhere. Relative roots resolve below `--project`. -### Opt-in: experience replay & dream rollouts +### Opt-in: experience replay, dream rollouts, and robustness probes -Two consolidation mechanisms, both default **off** (behavior is unchanged unless you -enable them). They strengthen the nightly update when your tasks have a clean -correctness signal; the validation gate still governs what ships. +These controls are default **off** (behavior is unchanged unless you enable +them). Replay and rollouts strengthen the training signal; adversarial probes +measure candidate robustness. The validation gate still governs what ships +unless explicit adversarial blocking adds a second rejection condition. | Config knob | Default | Effect | |---|---|---| | `dream_rollouts` | `1` | Run each task K times → learn from the good-vs-bad contrast (contrastive reflection). | | `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. | +| `dream_adversarial` | `0` | Score up to N harmless request-frame variants per real training task against each gate-eligible candidate. The factor is capped at 3 per task and 256 probes per candidate. | +| `dream_adversarial_blocking` | `false` | When `true`, reject a candidate if any probe drops beyond the configured margin. When `false`, surface the same evidence without changing the gate decision. | +| `dream_adversarial_margin` | `0.0` | Tolerated source-to-probe score drop in `[0, 1]` before a row is marked brittle. | + +Adversarial probes preserve the source reference and judge but change only the +request frame (for example, removing polite boilerplate or adding explicit +request delimiters). They are generated from real, underived training tasks +only. Recalled, already-synthetic, validation, and test tasks are excluded. +The report records every source/probe score and the exact perturbation that +failed. Probes are advisory first because any fixed robustness suite is an +incomplete proxy; enable blocking only after reviewing its behavior on your +task mix. Blocking mode fails closed if no eligible probe can be generated. +Each candidate adds one source rollout per eligible task plus N probe rollouts, +so token and latency cost grow with the number of real training tasks and the +selected factor. + +Example `~/.skillopt-sleep/config.json`: + +```json +{ + "dream_adversarial": 2, + "dream_adversarial_blocking": false, + "dream_adversarial_margin": 0.015 +} +``` ### Paired A/B evalkit diff --git a/docs/sleep/multi-skill-staging.md b/docs/sleep/multi-skill-staging.md index bb86b816..f0ebda6b 100644 --- a/docs/sleep/multi-skill-staging.md +++ b/docs/sleep/multi-skill-staging.md @@ -38,7 +38,9 @@ with a note in both report formats. They never fall back to the managed skill's document. The managed catch-all remains on `proposed_SKILL.md` and is not duplicated as a per-skill row. -Each group inherits `recall_k`, `dream_rollouts`, `dream_factor`, `edit_budget`, +Each group inherits `recall_k`, `dream_rollouts`, `dream_factor`, +`dream_adversarial`, `dream_adversarial_blocking`, +`dream_adversarial_margin`, `edit_budget`, `gate_mode`, `gate_metric`, `gate_mixed_weight`, `gate_no_regression`, and `evolve_skill`. Recalled archive tasks are restricted to the same skill hint, and shared memory is read-only during group runs. Consequently, diff --git a/skillopt_sleep/adversarial.py b/skillopt_sleep/adversarial.py new file mode 100644 index 00000000..f6869c3f --- /dev/null +++ b/skillopt_sleep/adversarial.py @@ -0,0 +1,224 @@ +"""Candidate-level robustness probes for SkillOpt-Sleep. + +The nightly gate measures a candidate on held-out tasks, but a candidate can +still be brittle to harmless changes in how a request is framed. This module +creates bounded, deterministic variants of *real training tasks* and compares +the candidate's score on each source task with its score on the corresponding +variant. + +Probes are evidence, not training examples: they never enter validation/test +splits and they never influence reflection. The caller decides whether a +flag is advisory or blocks a candidate. +""" +from __future__ import annotations + +import math +import re +from typing import Any, Dict, List, Sequence, Tuple + +from skillopt_sleep.backend import Backend +from skillopt_sleep.gate import select_gate_score +from skillopt_sleep.replay import replay_batch +from skillopt_sleep.types import ReplayResult, TaskRecord + +MAX_PROBES_PER_TASK = 3 +MAX_ADVERSARIAL_PROBES = 256 + + +def _normalize_split(value: str) -> str: + return {"replay": "train", "holdout": "val"}.get(value, value) + + +def _strip_polite_frame(intent: str) -> str: + """Remove only well-known request boilerplate; keep task semantics intact.""" + patterns = ( + r"(?is)^\s*please\s+", + r"(?is)^\s*(?:can|could|would)\s+you\s+", + r"(?is)^\s*i\s+(?:need|want)\s+you\s+to\s+", + ) + for pattern in patterns: + rewritten, count = re.subn(pattern, "", intent, count=1) + if count and rewritten.strip(): + rewritten = rewritten.strip() + return rewritten[:1].upper() + rewritten[1:] + return "" + + +def _variant_intents(intent: str) -> List[Tuple[str, str]]: + """Return conservative, deterministic surface variants in priority order.""" + raw = str(intent or "").strip() + if not raw: + return [] + variants: List[Tuple[str, str]] = [] + reframed = _strip_polite_frame(raw) + if reframed and reframed != raw: + variants.append(("request-frame", reframed)) + variants.extend(( + ( + "section-wrapper", + f"Task to complete:\n\n{raw}\n\nRespond to the task above.", + ), + ( + "delimiter-wrapper", + f"The request is between the markers.\n\n{raw}\n", + ), + ( + "boundary-shift", + f"Use the following request as the complete instruction:\n---\n{raw}\n---", + ), + )) + out: List[Tuple[str, str]] = [] + seen = {raw} + for kind, value in variants: + if value not in seen: + seen.add(value) + out.append((kind, value)) + return out + + +def generate_adversarial_probes( + tasks: Sequence[TaskRecord], + *, + factor: int = 1, +) -> List[TaskRecord]: + """Create bounded semantic-preserving probes from real TRAIN tasks only. + + ``factor`` is the maximum variants per source task and is capped at three. + Synthetic, recalled, validation, and test records are excluded so probes + cannot recycle held-out material or amplify already-derived tasks. + """ + if isinstance(factor, bool) or not isinstance(factor, int): + raise ValueError("adversarial probe factor must be an integer") + per_task = max(0, min(factor, MAX_PROBES_PER_TASK)) + if per_task == 0: + return [] + out: List[TaskRecord] = [] + source_ids: set[str] = set() + for task in tasks: + if len(out) >= MAX_ADVERSARIAL_PROBES: + break + if _normalize_split(task.split) != "train" or task.origin != "real": + continue + if task.derived_from or "recall" in (task.tags or []): + continue + if task.id in source_ids: + raise ValueError( + f"adversarial probe source ids must be unique: {task.id!r}" + ) + source_ids.add(task.id) + for kind, intent in _variant_intents(task.intent)[:per_task]: + if len(out) >= MAX_ADVERSARIAL_PROBES: + break + out.append(TaskRecord( + id=f"{task.id}_adversarial_{kind}", + project=task.project, + intent=intent, + context_excerpt=task.context_excerpt, + system=task.system, + attempted_solution=task.attempted_solution, + outcome=task.outcome, + reference_kind=task.reference_kind, + reference=task.reference, + judge=dict(task.judge), + tags=list(task.tags) + ["dream", "adversarial", f"probe:{kind}"], + source_sessions=list(task.source_sessions), + split="train", + origin="dream", + derived_from=task.id, + skill_hint=task.skill_hint, + )) + return out + + +def _score(result: ReplayResult, metric: str, mixed_weight: float) -> float | None: + value = select_gate_score(result.hard, result.soft, metric, mixed_weight) + return value if math.isfinite(value) else None + + +def evaluate_adversarial_probes( + backend: Backend, + tasks: Sequence[TaskRecord], + skill: str, + memory: str, + *, + factor: int = 1, + metric: str = "mixed", + mixed_weight: float = 0.5, + margin: float = 0.0, +) -> Dict[str, Any]: + """Score source/probe pairs and return a JSON-safe brittleness report. + + A row is flagged when its probe score falls more than ``margin`` below the + matching source score. Non-finite scores are invalid and fail closed. + """ + if isinstance(margin, bool) or not isinstance(margin, (int, float)): + raise ValueError("adversarial probe margin must be a finite number in [0, 1]") + numeric_margin = float(margin) + if not math.isfinite(numeric_margin) or not 0.0 <= numeric_margin <= 1.0: + raise ValueError("adversarial probe margin must be a finite number in [0, 1]") + + probes = generate_adversarial_probes(tasks, factor=factor) + source_by_id = { + task.id: task + for task in tasks + if _normalize_split(task.split) == "train" + and task.origin == "real" + and not task.derived_from + and "recall" not in (task.tags or []) + } + source_ids = list(dict.fromkeys(probe.derived_from for probe in probes)) + source_tasks = [source_by_id[source_id] for source_id in source_ids] + source_pairs = replay_batch(backend, source_tasks, skill, memory) + probe_pairs = replay_batch(backend, probes, skill, memory) + source_results = {task.id: result for task, result in source_pairs} + + rows: List[Dict[str, Any]] = [] + flagged = 0 + invalid = 0 + deltas: List[float] = [] + for probe, probe_result in probe_pairs: + source_result = source_results.get(probe.derived_from) + source_score = ( + _score(source_result, metric, mixed_weight) + if source_result is not None + else None + ) + probe_score = _score(probe_result, metric, mixed_weight) + valid = source_score is not None and probe_score is not None + delta = probe_score - source_score if valid else None + is_flagged = not valid or bool(delta is not None and delta < -numeric_margin) + if is_flagged: + flagged += 1 + if not valid: + invalid += 1 + if delta is not None: + deltas.append(delta) + kind = next( + (tag.removeprefix("probe:") for tag in probe.tags if tag.startswith("probe:")), + "unknown", + ) + rows.append({ + "source_task_id": probe.derived_from, + "probe_task_id": probe.id, + "probe_kind": kind, + "source_score": source_score, + "probe_score": probe_score, + "delta": delta, + "status": "invalid" if not valid else ("brittle" if is_flagged else "stable"), + }) + + n = len(rows) + return { + "enabled": True, + "factor": max(0, min(factor, MAX_PROBES_PER_TASK)), + "margin": numeric_margin, + "n_sources": len(source_tasks), + "n_probes": n, + "n_flagged": flagged, + "n_invalid": invalid, + "brittleness_rate": (flagged / n) if n else 0.0, + "worst_delta": min(deltas) if deltas else None, + "conclusive": n > 0, + "flagged": flagged > 0, + "rows": rows, + } diff --git a/skillopt_sleep/config.py b/skillopt_sleep/config.py index d4a008d1..1d73caad 100644 --- a/skillopt_sleep/config.py +++ b/skillopt_sleep/config.py @@ -70,6 +70,9 @@ # ── dream + recall (opt-in; defaults reproduce the prior single-shot loop) ─ "dream_rollouts": 1, # >1 => multi-rollout contrastive reflection per task "dream_factor": 0, # >0 => add N synthetic variants of each task to the dream + "dream_adversarial": 0, # >0 => score N robustness probes per real train task + "dream_adversarial_blocking": False, # reject flagged candidates instead of advisory-only + "dream_adversarial_margin": 0.0, # tolerated source->probe score drop in [0, 1] "recall_k": 0, # >0 => recall the K most-similar past tasks into the dream "evolve_memory": True, # consolidate CLAUDE.md "evolve_skill": True, # consolidate the managed SKILL.md diff --git a/skillopt_sleep/consolidate.py b/skillopt_sleep/consolidate.py index 5b80b4a9..8952d10b 100644 --- a/skillopt_sleep/consolidate.py +++ b/skillopt_sleep/consolidate.py @@ -12,6 +12,7 @@ from dataclasses import dataclass, field from typing import List, Tuple +from skillopt_sleep.adversarial import evaluate_adversarial_probes from skillopt_sleep.backend import Backend # Self-contained validation gate (vendored from SkillOpt; zero dependency on the @@ -187,6 +188,9 @@ def consolidate( gate_metric: str = "mixed", gate_mixed_weight: float = 0.5, gate_no_regression: bool = False, + dream_adversarial: int = 0, + dream_adversarial_blocking: bool = False, + dream_adversarial_margin: float = 0.0, gate_mode: str = "on", # "on" (hard/soft per gate_metric) | "off" (greedy) rollouts_k: int = 1, # >1 => multi-rollout contrastive reflection evolve_skill: bool = True, @@ -202,6 +206,24 @@ def consolidate( Skill and memory are evolved in sequence (skill first if both enabled). """ + if isinstance(dream_adversarial, bool) or not isinstance( + dream_adversarial, int + ): + raise ValueError("dream_adversarial must be an integer") + if dream_adversarial > 0: + if not isinstance(dream_adversarial_blocking, bool): + raise ValueError("dream_adversarial_blocking must be a boolean") + if isinstance(dream_adversarial_margin, bool) or not isinstance( + dream_adversarial_margin, (int, float) + ): + raise ValueError( + "dream_adversarial_margin must be a finite number in [0, 1]" + ) + margin = float(dream_adversarial_margin) + if not math.isfinite(margin) or not 0.0 <= margin <= 1.0: + raise ValueError( + "dream_adversarial_margin must be a finite number in [0, 1]" + ) from skillopt_sleep import evidence as evlog ev = evlog.get(backend) train_tasks, val_tasks, holdout_leaked = _split(tasks) @@ -261,16 +283,69 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: n_edits=len(unmatched), edits=_edits_payload(unmatched)) if not applied: return doc + trial_skill = new_doc if which == "skill" else cand_skill + trial_memory = new_doc if which == "memory" else cand_memory + + def _probe_candidate() -> tuple[dict | None, bool]: + if dream_adversarial <= 0: + return None, False + evlog.set_phase(backend, f"adversarial_probe:{which}") + probe = evaluate_adversarial_probes( + backend, + train_tasks, + trial_skill, + trial_memory, + factor=dream_adversarial, + metric=gate_metric, + mixed_weight=gate_mixed_weight, + margin=dream_adversarial_margin, + ) + blocked = bool( + dream_adversarial_blocking + and (probe["flagged"] or not probe["conclusive"]) + ) + probe["blocking"] = bool(dream_adversarial_blocking) + probe["blocked"] = blocked + probe["block_reason"] = ( + "inconclusive_no_probes" + if blocked and not probe["conclusive"] + else ("brittle_score_drop" if blocked else "") + ) + if ev is not None: + ev.log( + "adversarial", + "candidate_probe", + target=which, + **probe, + ) + return probe, blocked + # gate OFF: accept greedily with NO val scoring (the daily-use path) if gate_off: - all_applied.extend(applied) + probe, blocked_by_adversarial = _probe_candidate() + accepted = not blocked_by_adversarial + if accepted: + all_applied.extend(applied) + else: + all_rejected.extend(applied) + if probe is not None: + gate_trials.append({ + "target": which, + "baseline_score": None, + "candidate_score": None, + "accepted": accepted, + "blocked_by_regression": False, + "blocked_by_adversarial": blocked_by_adversarial, + "task_deltas": [], + "adversarial_probe": probe, + }) if ev is not None: ev.log("gate", "trial", target=which, mode="greedy", - accepted=True, n_edits=len(applied)) - return new_doc + accepted=accepted, + blocked_by_adversarial=blocked_by_adversarial, + n_edits=len(applied)) + return new_doc if accepted else doc # gate ON: score the candidate on the VAL slice, keep only if it improves - trial_skill = new_doc if which == "skill" else cand_skill - trial_memory = new_doc if which == "memory" else cand_memory evlog.set_phase(backend, f"gate_trial:{which}") pairs = replay_batch(backend, val_tasks, trial_skill, trial_memory) h, s = aggregate_scores(pairs) @@ -283,22 +358,40 @@ def _gate_apply(doc: str, edits: List[EditRecord], which: str) -> str: and any(row["status"] == "regressed" for row in task_deltas) ) trial_base_score = base_score - improved = cand_score > base_score and not blocked_by_regression - gate_trials.append({ + gate_improved = cand_score > base_score and not blocked_by_regression + probe = None + blocked_by_adversarial = False + if gate_improved: + probe, blocked_by_adversarial = _probe_candidate() + improved = gate_improved and not blocked_by_adversarial + trial_record = { "target": which, "baseline_score": _finite_score(trial_base_score), "candidate_score": _finite_score(cand_score), "accepted": improved, "blocked_by_regression": blocked_by_regression, "task_deltas": task_deltas, - }) + } + if dream_adversarial > 0: + trial_record["blocked_by_adversarial"] = blocked_by_adversarial + trial_record["adversarial_probe"] = probe + gate_trials.append(trial_record) if ev is not None: - ev.log("gate", "trial", target=which, mode="gated", - baseline_score=trial_base_score, cand_hard=h, cand_soft=s, - cand_score=cand_score, accepted=improved, - blocked_by_regression=blocked_by_regression, - task_deltas=task_deltas, - n_edits=len(applied)) + trial_evidence = { + "target": which, + "mode": "gated", + "baseline_score": trial_base_score, + "cand_hard": h, + "cand_soft": s, + "cand_score": cand_score, + "accepted": improved, + "blocked_by_regression": blocked_by_regression, + "task_deltas": task_deltas, + "n_edits": len(applied), + } + if dream_adversarial > 0: + trial_evidence["blocked_by_adversarial"] = blocked_by_adversarial + ev.log("gate", "trial", **trial_evidence) if improved: base_score = cand_score current_pairs = pairs diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py index f1d7bf02..85444220 100644 --- a/skillopt_sleep/cycle.py +++ b/skillopt_sleep/cycle.py @@ -27,7 +27,6 @@ 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, @@ -36,6 +35,7 @@ consolidate_groups, skill_group_reports, ) +from skillopt_sleep.replay import aggregate_scores, replay_batch from skillopt_sleep.skill_resolver import resolve_skill, skill_search_roots from skillopt_sleep.staging import ( SkillProposal, @@ -317,6 +317,18 @@ def _render_report_md(report: SleepReport, cfg: SleepConfig) -> str: f"- tokens used: {report.tokens_used}", "", ] + if int(cfg.get("dream_adversarial", 0) or 0) > 0: + mode = ( + "blocking" + if cfg.get("dream_adversarial_blocking", False) + else "advisory" + ) + lines.insert( + -1, + f"- adversarial dream probes: {mode} " + f"(factor={int(cfg.get('dream_adversarial', 0) or 0)}, " + f"margin={_report_score(cfg.get('dream_adversarial_margin', 0.0))})", + ) gate_on = str(cfg.get("gate_mode", "on")).strip().lower() not in { "off", "none", "false", "greedy", } @@ -341,9 +353,14 @@ def _render_report_md(report: SleepReport, cfg: SleepConfig) -> str: target = _markdown_table_text(trial.get("target", "candidate")) accepted = bool(trial.get("accepted", False)) blocked = bool(trial.get("blocked_by_regression", False)) + adversarial_blocked = bool( + trial.get("blocked_by_adversarial", False) + ) decision = "accepted" if accepted else "rejected" if blocked: decision += " (task regression)" + if adversarial_blocked: + decision += " (brittle under adversarial probes)" baseline = _report_score(trial.get("baseline_score")) candidate = _report_score(trial.get("candidate_score")) lines.append( @@ -366,6 +383,36 @@ def _render_report_md(report: SleepReport, cfg: SleepConfig) -> str: f"| {candidate_score} | {status} |" ) lines.append("") + probe = trial.get("adversarial_probe") + if isinstance(probe, dict): + mode = "blocking" if probe.get("blocking") else "advisory" + lines.append( + f"Adversarial probes ({mode}): " + f"{int(probe.get('n_flagged', 0) or 0)} flagged / " + f"{int(probe.get('n_probes', 0) or 0)} total; " + f"worst delta {_report_score(probe.get('worst_delta'))}." + ) + if not probe.get("conclusive", False): + lines.append( + "No eligible probe was generated; blocking mode fails " + "closed for this candidate." + ) + lines.append("") + lines.append("| Source | Probe | Variant | Source | Probe | Delta | Status |") + lines.append("|---|---|---|---:|---:|---:|---|") + for row in probe.get("rows", []): + source_id = _markdown_table_text(row.get("source_task_id", "")) + probe_id = _markdown_table_text(row.get("probe_task_id", "")) + kind = _markdown_table_text(row.get("probe_kind", "")) + source_score = _report_score(row.get("source_score")) + probe_score = _report_score(row.get("probe_score")) + delta = _report_score(row.get("delta")) + status = _markdown_table_text(row.get("status", "")) + lines.append( + f"| `{source_id}` | `{probe_id}` | {kind} | " + f"{source_score} | {probe_score} | {delta} | {status} |" + ) + lines.append("") if report.edits: lines.append("## Accepted edits") for e in report.edits: @@ -698,6 +745,16 @@ def run_sleep_cycle( "dream_rollouts", "dream_factor", "recall_k", "max_tasks_per_night", "lookback_hours", "llm_mine", "evolve_skill", "evolve_memory")} + if int(cfg.get("dream_adversarial", 0) or 0) > 0: + cycle_config.update({ + "dream_adversarial": cfg.get("dream_adversarial", 0), + "dream_adversarial_blocking": cfg.get( + "dream_adversarial_blocking", False + ), + "dream_adversarial_margin": cfg.get( + "dream_adversarial_margin", 0.0 + ), + }) cycle_config["opencode_tool_replay"] = ( cfg.get("opencode_tool_replay", False) is True ) @@ -844,8 +901,9 @@ def run_sleep_cycle( # ── 3+4. replay + consolidate (gate), with opt-in dream + recall ────── # recall pulls similar past tasks from the persisted archive; dream_rollouts - # / dream_factor enrich the training signal. With the defaults (recall_k=0, - # dream_rollouts=1, dream_factor=0) this is exactly the prior single-shot + # / dream_factor enrich the training signal. dream_adversarial instead + # scores candidate robustness without entering reflection or held-out data. + # With every option disabled this is exactly the prior single-shot # consolidate — behavior is unchanged unless the user opts in. _progress(cfg, "consolidate start") recall_k = int(cfg.get("recall_k", 0) or 0) @@ -859,6 +917,11 @@ def run_sleep_cycle( recall_k=recall_k, dream_rollouts=int(cfg.get("dream_rollouts", 1) or 1), dream_factor=int(cfg.get("dream_factor", 0) or 0), + dream_adversarial=cfg.get("dream_adversarial", 0), + dream_adversarial_blocking=cfg.get( + "dream_adversarial_blocking", False + ), + dream_adversarial_margin=cfg.get("dream_adversarial_margin", 0.0), edit_budget=cfg.get("edit_budget", 4), gate_metric=cfg.get("gate_metric", "mixed"), gate_mixed_weight=cfg.get("gate_mixed_weight", 0.5), @@ -953,6 +1016,13 @@ def run_sleep_cycle( recall_k=recall_k, dream_rollouts=int(cfg.get("dream_rollouts", 1) or 1), dream_factor=int(cfg.get("dream_factor", 0) or 0), + dream_adversarial=cfg.get("dream_adversarial", 0), + dream_adversarial_blocking=cfg.get( + "dream_adversarial_blocking", False + ), + dream_adversarial_margin=cfg.get( + "dream_adversarial_margin", 0.0 + ), edit_budget=cfg.get("edit_budget", 4), gate_metric=cfg.get("gate_metric", "mixed"), gate_mixed_weight=cfg.get("gate_mixed_weight", 0.5), @@ -1039,6 +1109,15 @@ def run_sleep_cycle( ), "gate_mode": cfg.get("gate_mode"), "gate_no_regression": cfg.get("gate_no_regression", False), + **({ + "dream_adversarial": cfg.get("dream_adversarial", 0), + "dream_adversarial_blocking": cfg.get( + "dream_adversarial_blocking", False + ), + "dream_adversarial_margin": cfg.get( + "dream_adversarial_margin", 0.0 + ), + } if int(cfg.get("dream_adversarial", 0) or 0) > 0 else {}), "n_tasks": len(tasks), "baseline_score": result.baseline_score, "candidate_score": result.candidate_score, diff --git a/skillopt_sleep/dream.py b/skillopt_sleep/dream.py index 9906e07d..b0c3fd75 100644 --- a/skillopt_sleep/dream.py +++ b/skillopt_sleep/dream.py @@ -1,6 +1,6 @@ """SkillOpt-Sleep — dream + associative recall for nightly consolidation. -Two opt-in mechanisms (both default OFF, so the cycle is unchanged unless the +Three opt-in mechanisms (all default OFF, so the cycle is unchanged unless the user enables them) that the deployment experiments validated: * dream rollouts — run each task K times and learn from the good-vs-bad @@ -8,6 +8,9 @@ * associative recall — each night, pull the K past tasks most similar to tonight's new ones into the dream (set ``recall_k > 0``). Replays relevant experience without re-running the whole history. + * adversarial probes — replay harmless surface variants against a candidate + before adoption (set ``dream_adversarial > 0``). Advisory by default; + blocking requires an explicit second switch. ``dream_consolidate`` wires recall + synthetic augmentation + multi-rollout consolidation and is called by BOTH the shipped plugin cycle and the benchmark @@ -130,6 +133,9 @@ def dream_consolidate( gate_metric: str = "mixed", gate_mixed_weight: float = 0.5, gate_no_regression: bool = False, + dream_adversarial: int = 0, + dream_adversarial_blocking: bool = False, + dream_adversarial_margin: float = 0.0, gate_mode: str = "on", evolve_skill: bool = True, evolve_memory: bool = True, @@ -140,8 +146,9 @@ def dream_consolidate( ``tasks`` is the split-tagged pool for tonight (train + val); recall and augmentation only enlarge the TRAIN split, so the val slice the gate scores - on is never polluted. With ``recall_k=0`` and ``dream_rollouts=1`` (the - defaults) this is exactly the previous single-shot ``consolidate``. + on is never polluted. Adversarial probes are scored separately and never + enter reflection or a held-out split. With every dream option disabled this + is exactly the previous single-shot ``consolidate``. """ train = [t for t in tasks if t.split == "train"] enlarged = list(tasks) @@ -162,7 +169,11 @@ def dream_consolidate( backend, enlarged, skill, memory, edit_budget=edit_budget, gate_metric=gate_metric, gate_mixed_weight=gate_mixed_weight, - gate_no_regression=gate_no_regression, gate_mode=gate_mode, + gate_no_regression=gate_no_regression, + dream_adversarial=dream_adversarial, + dream_adversarial_blocking=dream_adversarial_blocking, + dream_adversarial_margin=dream_adversarial_margin, + gate_mode=gate_mode, rollouts_k=dream_rollouts, evolve_skill=evolve_skill, evolve_memory=evolve_memory, night=night, ) diff --git a/tests/test_adversarial_dream.py b/tests/test_adversarial_dream.py new file mode 100644 index 00000000..4e07ed28 --- /dev/null +++ b/tests/test_adversarial_dream.py @@ -0,0 +1,429 @@ +"""Adversarial dream probes reject brittle candidate rules before adoption.""" +from __future__ import annotations + +import json +import os +from dataclasses import asdict + +import pytest + +from skillopt_sleep.adversarial import ( + MAX_ADVERSARIAL_PROBES, + evaluate_adversarial_probes, + generate_adversarial_probes, +) +from skillopt_sleep.backend import DualBackend, MockBackend +from skillopt_sleep.config import DEFAULTS, load_config +from skillopt_sleep.consolidate import consolidate +from skillopt_sleep.cycle import run_sleep_cycle +from skillopt_sleep.types import EditRecord, TaskRecord + + +def _task( + task_id: str, + *, + intent: str = "Please return ok", + split: str = "train", + origin: str = "real", + derived_from: str = "", +) -> TaskRecord: + return TaskRecord( + id=task_id, + project="/project", + intent=intent, + context_excerpt="context", + system="system", + attempted_solution="prior", + outcome="fail", + reference_kind="exact", + reference="ok", + judge={"kind": "exact"}, + tags=["fixture"], + source_sessions=["session-1"], + split=split, + origin=origin, + derived_from=derived_from, + skill_hint="demo-skill", + ) + + +class _CandidateBackend(MockBackend): + """The planted rule works only on the harvested task's exact surface.""" + + RULE = "Use the planted literal-only rule." + HARVESTED_INTENTS = {"Please return ok", "Say ok"} + + def reflect( + self, + failures, + successes, + skill, + memory, + *, + edit_budget, + evolve_skill, + evolve_memory, + ): + if self.RULE in f"{skill}\n{memory}": + return [] + return [ + EditRecord( + target="skill" if evolve_skill else "memory", + op="add", + content=self.RULE, + rationale="planted brittle candidate", + ) + ] + + def attempt(self, task, skill, memory, sample_id=0): + if self.RULE not in f"{skill}\n{memory}": + return "wrong" + return task.reference if task.intent in self.HARVESTED_INTENTS else "wrong" + + +class _RobustCandidateBackend(_CandidateBackend): + def attempt(self, task, skill, memory, sample_id=0): + if self.RULE not in f"{skill}\n{memory}": + return "wrong" + return task.reference + + +class _CountingRobustBackend(_RobustCandidateBackend): + def __init__(self) -> None: + self.attempt_calls = 0 + + def attempt(self, task, skill, memory, sample_id=0): + self.attempt_calls += 1 + return super().attempt(task, skill, memory, sample_id) + + +class _NonFiniteProbeBackend(_RobustCandidateBackend): + def judge(self, task, response): + if task.origin == "dream": + return float("nan"), 0.0, "invalid probe score" + return super().judge(task, response) + + +class _CountingRoleBackend(_RobustCandidateBackend): + def __init__(self, name: str) -> None: + self.name = name + self.attempt_calls = 0 + self.judge_calls = 0 + + def attempt(self, task, skill, memory, sample_id=0): + self.attempt_calls += 1 + return super().attempt(task, skill, memory, sample_id) + + def judge(self, task, response): + self.judge_calls += 1 + return super().judge(task, response) + + +def _candidate_tasks() -> list[TaskRecord]: + return [_task("train"), _task("val", intent="Say ok", split="val")] + + +def test_probe_generation_uses_only_real_underived_train_tasks() -> None: + source = _task("source") + tasks = [ + source, + _task("val", split="val"), + _task("test", split="test"), + _task("dream", origin="dream", derived_from="source"), + _task("recall", derived_from="old"), + ] + + probes = generate_adversarial_probes(tasks, factor=3) + + assert len(probes) == 3 + assert {probe.derived_from for probe in probes} == {"source"} + assert len({probe.id for probe in probes}) == 3 + assert all(probe.split == "train" and probe.origin == "dream" for probe in probes) + assert all(probe.intent != source.intent for probe in probes) + assert all("adversarial" in probe.tags for probe in probes) + assert all(probe.reference == source.reference for probe in probes) + assert all(probe.judge == source.judge for probe in probes) + assert all(probe.system == source.system for probe in probes) + assert all(probe.source_sessions == source.source_sessions for probe in probes) + assert probes[0].intent == "Return ok" + + +def test_probe_generation_is_bounded_and_rejects_ambiguous_factor_types() -> None: + tasks = [_task(f"task-{index}") for index in range(100)] + + probes = generate_adversarial_probes(tasks, factor=99) + + assert len(probes) == MAX_ADVERSARIAL_PROBES + with pytest.raises(ValueError, match="must be an integer"): + generate_adversarial_probes(tasks, factor=True) + with pytest.raises(ValueError, match="source ids must be unique"): + generate_adversarial_probes([_task("duplicate"), _task("duplicate")]) + + +def test_probe_report_flags_a_planted_literal_surface_rule() -> None: + report = evaluate_adversarial_probes( + _CandidateBackend(), + [_task("train")], + _CandidateBackend.RULE, + "", + factor=2, + ) + + assert report["n_sources"] == 1 + assert report["n_probes"] == 2 + assert report["n_flagged"] == 2 + assert report["brittleness_rate"] == 1.0 + assert report["worst_delta"] == -1.0 + assert {row["status"] for row in report["rows"]} == {"brittle"} + + +def test_probe_report_keeps_a_surface_robust_rule_stable() -> None: + report = evaluate_adversarial_probes( + _RobustCandidateBackend(), + [_task("train")], + _CandidateBackend.RULE, + "", + factor=3, + ) + + assert report["n_flagged"] == 0 + assert report["flagged"] is False + assert report["worst_delta"] == 0.0 + assert {row["status"] for row in report["rows"]} == {"stable"} + + +def test_dual_backend_probes_route_attempts_and_exact_judging_to_target() -> None: + target = _CountingRoleBackend("target") + optimizer = _CountingRoleBackend("optimizer") + dual = DualBackend(target=target, optimizer=optimizer) + + report = evaluate_adversarial_probes( + dual, + [_task("train")], + _CandidateBackend.RULE, + "", + ) + + assert report["flagged"] is False + assert target.attempt_calls == 2 # source + one probe + assert target.judge_calls == 2 + assert optimizer.attempt_calls == 0 + assert optimizer.judge_calls == 0 + + +def test_non_finite_probe_score_is_json_safe_and_fails_closed() -> None: + report = evaluate_adversarial_probes( + _NonFiniteProbeBackend(), + [_task("train")], + _CandidateBackend.RULE, + "", + ) + + assert report["flagged"] is True + assert report["n_invalid"] == 1 + assert report["rows"][0]["probe_score"] is None + assert report["rows"][0]["status"] == "invalid" + json.dumps(report, allow_nan=False) + + +@pytest.mark.parametrize("margin", [True, -0.1, 1.1, float("nan"), "0.1"]) +def test_probe_margin_rejects_non_finite_or_out_of_range_values(margin) -> None: + with pytest.raises(ValueError, match="finite number"): + evaluate_adversarial_probes( + _RobustCandidateBackend(), + [_task("train")], + _CandidateBackend.RULE, + "", + margin=margin, + ) + + +@pytest.mark.parametrize( + ("overrides", "message"), + [ + ({"dream_adversarial": True}, "must be an integer"), + ({"dream_adversarial": 1.5}, "must be an integer"), + ( + { + "dream_adversarial": 1, + "dream_adversarial_blocking": "false", + }, + "must be a boolean", + ), + ( + {"dream_adversarial": 1, "dream_adversarial_margin": float("inf")}, + "finite number", + ), + ], +) +def test_consolidate_rejects_ambiguous_adversarial_config(overrides, message) -> None: + with pytest.raises(ValueError, match=message): + consolidate( + _RobustCandidateBackend(), + _candidate_tasks(), + "# skill\n", + "", + evolve_memory=False, + **overrides, + ) + + +def test_advisory_probe_surfaces_brittleness_without_changing_gate_decision() -> None: + result = consolidate( + _CandidateBackend(), + _candidate_tasks(), + "# skill\n", + "", + evolve_memory=False, + dream_adversarial=2, + dream_adversarial_blocking=False, + ) + + assert result.accepted is True + trial = result.gate_trials[0] + assert trial["accepted"] is True + assert trial["blocked_by_adversarial"] is False + assert trial["adversarial_probe"]["flagged"] is True + assert trial["adversarial_probe"]["blocking"] is False + + +def test_advisory_probe_is_held_out_result_equivalent_for_robust_candidate() -> None: + off = consolidate( + _RobustCandidateBackend(), + _candidate_tasks(), + "# skill\n", + "", + evolve_memory=False, + ) + on = consolidate( + _RobustCandidateBackend(), + _candidate_tasks(), + "# skill\n", + "", + evolve_memory=False, + dream_adversarial=3, + dream_adversarial_blocking=False, + ) + + assert on.accepted == off.accepted is True + assert on.gate_action == off.gate_action + assert on.baseline_score == off.baseline_score + assert on.candidate_score == off.candidate_score + assert on.new_skill == off.new_skill + assert on.new_memory == off.new_memory + assert on.gate_trials[0]["adversarial_probe"]["n_flagged"] == 0 + + +def test_blocking_probe_rejects_candidate_that_passed_the_held_out_gate() -> None: + result = consolidate( + _CandidateBackend(), + _candidate_tasks(), + "# skill\n", + "", + evolve_memory=False, + dream_adversarial=2, + dream_adversarial_blocking=True, + ) + + assert result.accepted is False + assert result.applied_edits == [] + assert [edit.content for edit in result.rejected_edits] == [_CandidateBackend.RULE] + trial = result.gate_trials[0] + assert trial["candidate_score"] == 1.0 + assert trial["accepted"] is False + assert trial["blocked_by_adversarial"] is True + assert trial["adversarial_probe"]["blocked"] is True + + +def test_blocking_probe_fails_closed_when_no_eligible_variant_exists() -> None: + result = consolidate( + _CandidateBackend(), + [_task("train", intent=""), _task("val", intent="Say ok", split="val")], + "# skill\n", + "", + evolve_memory=False, + dream_adversarial=1, + dream_adversarial_blocking=True, + ) + + assert result.accepted is False + probe = result.gate_trials[0]["adversarial_probe"] + assert probe["conclusive"] is False + assert probe["blocked"] is True + assert probe["block_reason"] == "inconclusive_no_probes" + + +def test_default_off_is_result_identical_to_explicit_zero() -> None: + default_backend = _CountingRobustBackend() + explicit_backend = _CountingRobustBackend() + default = consolidate( + default_backend, + _candidate_tasks(), + "# skill\n", + "", + evolve_memory=False, + ) + explicit = consolidate( + explicit_backend, + _candidate_tasks(), + "# skill\n", + "", + evolve_memory=False, + dream_adversarial=0, + dream_adversarial_blocking=False, + dream_adversarial_margin=0.0, + ) + + assert asdict(default) == asdict(explicit) + assert default_backend.attempt_calls == explicit_backend.attempt_calls == 4 + assert "adversarial_probe" not in default.gate_trials[0] + assert "blocked_by_adversarial" not in default.gate_trials[0] + assert DEFAULTS["dream_adversarial"] == 0 + assert DEFAULTS["dream_adversarial_blocking"] is False + assert DEFAULTS["dream_adversarial_margin"] == 0.0 + assert load_config().get("dream_adversarial") == 0 + + +def test_cycle_persists_advisory_probe_evidence_in_review_artifacts(tmp_path) -> None: + project = tmp_path / "project" + project.mkdir() + config = load_config( + invoked_project=str(project), + projects="invoked", + backend="mock", + state_dir=str(tmp_path / "state"), + claude_home=str(tmp_path / ".claude"), + evolve_memory=False, + dream_adversarial=2, + dream_adversarial_blocking=False, + dream_adversarial_margin=0.015, + auto_adopt=False, + ) + + tasks = _candidate_tasks() + tasks[0].id = "train|