diff --git a/docs/sleep/README.md b/docs/sleep/README.md
index f47556eeb..f9c70aaf5 100644
--- a/docs/sleep/README.md
+++ b/docs/sleep/README.md
@@ -308,7 +308,9 @@ 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`,
+`dream_adversarial_rollouts`, `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 +329,66 @@ 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 whose brittleness is candidate-introduced under the baseline-relative rule below. When `false`, surface the same evidence without changing the gate decision. Requires `dream_adversarial_rollouts >= 2`. |
+| `dream_adversarial_margin` | `0.0` | Tolerated worsening of the candidate gap relative to the baseline gap, in `[0, 1]`, before a row is marked brittle. Calibrate it on your own task mix before enabling blocking. |
+| `dream_adversarial_rollouts` | `1` | Repeated samples per task and arm (capped at 8). Blocking requires at least 2 so one stochastic sample can never reject a candidate. |
+
+Adversarial probes preserve the source reference and judge but change only the
+request frame (for example, removing explicitly politeness-marked boilerplate
+or adding request delimiters). They are generated from real, underived
+training tasks only. Recalled, already-synthetic, validation, and test tasks
+are excluded.
+
+The decision is **baseline-relative** so pre-existing frame sensitivity never
+flags a candidate: every source/probe pair is scored under both the current
+(baseline) documents and the candidate documents, each score is the mean of
+`dream_adversarial_rollouts` repeated samples, and a row is brittle only when
+the candidate's probe-minus-source gap worsens beyond the margin relative to
+the baseline gap AND the worsening holds in a strict majority of rollout
+indices. All four aggregated scores and the per-rollout samples are retained
+in the evidence so the decision is auditable. Any non-finite score fails
+closed.
+
+Repeated rollouts are only evidence when they are distinct samples. Every
+replay route forwards the rollout's `sample_id`: the text route and the
+inherited `TOOL_CALL:` marker fallback salt the attempt cache key with it, and
+the real tool-loop backends start a fresh, uncached agent process per call.
+A custom backend whose `attempt` or `attempt_with_tools` does not accept
+`sample_id` cannot provide that guarantee, so with more than one rollout its
+rows are reported as `inconclusive` and not replayed. Inconclusive rows never
+flag a candidate; if no conclusive row remains, blocking mode fails closed and
+names the unsupported route in `block_reason`.
+
+Probes are advisory first because any fixed robustness suite is an incomplete
+proxy; enable blocking only after reviewing the advisory evidence and
+calibrating the margin on your task mix. Blocking mode fails closed if no
+eligible probe can be generated. The replay cost per gate-eligible candidate
+is `rollouts * 2 * (sources + probes)` over the conclusive rows, so token and
+latency cost grow with
+the number of real training tasks, the factor, and the rollout count.
+
+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 bb86b8169..77faf9f87 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`, `dream_adversarial_rollouts`, `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 000000000..57feab313
--- /dev/null
+++ b/skillopt_sleep/adversarial.py
@@ -0,0 +1,397 @@
+"""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 repeated_samples_distinct, replay_one
+from skillopt_sleep.types import ReplayResult, TaskRecord
+
+MAX_PROBES_PER_TASK = 3
+MAX_ADVERSARIAL_PROBES = 256
+MAX_PROBE_ROLLOUTS = 8
+# Blocking decisions need repeated rollouts so a single stochastic sample can
+# never reject a candidate on its own; advisory runs may use one rollout.
+MIN_BLOCKING_ROLLOUTS = 2
+
+
+def _normalize_split(value: str) -> str:
+    return {"replay": "train", "holdout": "val"}.get(value, value)
+
+
+def _strip_polite_frame(intent: str) -> str:
+    """Reframe explicitly politeness-marked requests, and nothing else.
+
+    Semantic-preservation contract: a transformation is emitted only when the
+    removed prefix is an unambiguous request marker, so removal cannot change
+    what is being asked:
+
+    * a leading ``please`` (a pure politeness marker), and
+    * a leading ``can/could/would you please`` (a modal question that the
+      politeness marker disambiguates as a request; the trailing question
+      mark, if any, becomes a period because the reframed text is the same
+      request in imperative form).
+
+    Bare modal questions (``Can you swim?``, ``Would you like tea?``) are
+    never reframed: without the politeness marker they may ask about ability,
+    permission, or desire, and stripping the modal changes the meaning. The
+    same applies to first-person desire framings (``I want you to ...``),
+    which earlier revisions stripped and this contract deliberately drops.
+    """
+    modal_request = r"(?is)^\s*(?:can|could|would)\s+you\s+please\s+"
+    plain_please = r"(?is)^\s*please\s+"
+    for pattern, reframed_request in ((modal_request, True), (plain_please, False)):
+        rewritten, count = re.subn(pattern, "", intent, count=1)
+        if not count or not rewritten.strip():
+            continue
+        rewritten = rewritten.strip()
+        if not rewritten[:1].isalpha():
+            return ""
+        if reframed_request and rewritten.endswith("?"):
+            rewritten = rewritten[:-1].rstrip()
+            if not rewritten or not rewritten[:1].isalpha():
+                return ""
+            rewritten += "."
+        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<request>\n{raw}\n</request>",
+        ),
+        (
+            "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 _rollout_scores(
+    backend: Backend,
+    task: TaskRecord,
+    skill: str,
+    memory: str,
+    *,
+    metric: str,
+    mixed_weight: float,
+    rollouts: int,
+) -> List[float | None]:
+    """Score one task ``rollouts`` times under one document pair.
+
+    Each rollout uses a distinct ``sample_id`` so caching backends produce
+    genuinely repeated samples instead of collapsing to one response.
+    """
+    scores: List[float | None] = []
+    for sample_id in range(rollouts):
+        result = replay_one(backend, task, skill, memory, sample_id=sample_id)
+        scores.append(_score(result, metric, mixed_weight))
+    return scores
+
+
+def _mean(values: Sequence[float]) -> float:
+    return sum(values) / len(values)
+
+
+def evaluate_adversarial_probes(
+    backend: Backend,
+    tasks: Sequence[TaskRecord],
+    skill: str,
+    memory: str,
+    *,
+    baseline_skill: str,
+    baseline_memory: str,
+    factor: int = 1,
+    metric: str = "mixed",
+    mixed_weight: float = 0.5,
+    margin: float = 0.0,
+    rollouts: int = 1,
+) -> Dict[str, Any]:
+    """Score identical source/probe pairs under the BASELINE and the CANDIDATE
+    documents and report candidate-introduced brittleness.
+
+    Decision rule (documented so the evidence is auditable):
+
+    * every task in every arm is replayed ``rollouts`` times and the arm's
+      score for that task is the MEAN of those rollouts;
+    * per row, ``gap = probe_score - source_score`` is computed for both the
+      baseline arm and the candidate arm, and
+      ``gap_change = candidate_gap - baseline_gap``;
+    * a row is brittle only when ``gap_change < -margin`` AND the per-index
+      paired worsening holds in a strict majority of rollout indices;
+    * any non-finite score in any arm marks the row invalid, which fails
+      closed: invalid rows count as flagged;
+    * with ``rollouts > 1``, a row whose route cannot obtain distinct
+      repeated samples (see ``repeated_samples_distinct``) is not replayed
+      and is marked ``inconclusive``: repeated copies of one response would
+      claim a confidence the evidence does not have. Inconclusive rows are
+      never flagged, and when no conclusive row remains the result is
+      ``conclusive=False``, which blocking mode treats as a fail-closed
+      rejection.
+
+    Frame sensitivity already present under the baseline documents therefore
+    never flags a candidate; only the change the candidate introduces does.
+    All four aggregated scores and the per-rollout samples are retained per
+    row so the decision can be audited from the evidence alone. The total
+    replay cost is ``rollouts * 2 * (sources + probes)`` over the conclusive
+    rows and the sources they reference.
+    """
+    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]")
+    if isinstance(rollouts, bool) or not isinstance(rollouts, int):
+        raise ValueError(
+            f"adversarial probe rollouts must be an integer in [1, {MAX_PROBE_ROLLOUTS}]"
+        )
+    if not 1 <= rollouts <= MAX_PROBE_ROLLOUTS:
+        raise ValueError(
+            f"adversarial probe rollouts must be an integer in [1, {MAX_PROBE_ROLLOUTS}]"
+        )
+
+    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))
+    # A single rollout makes no independence claim. Repeated rollouts must
+    # reach the model as distinct samples on BOTH tasks of a pair; a probe
+    # keeps its source's judge, so both normally share one route.
+    distinct = {
+        probe.id: rollouts == 1 or (
+            repeated_samples_distinct(backend, source_by_id[probe.derived_from])
+            and repeated_samples_distinct(backend, probe)
+        )
+        for probe in probes
+    }
+    scored_probes = [probe for probe in probes if distinct[probe.id]]
+    scored_source_ids = list(
+        dict.fromkeys(probe.derived_from for probe in scored_probes)
+    )
+    arms = {
+        "baseline": (baseline_skill, baseline_memory),
+        "candidate": (skill, memory),
+    }
+    source_scores: Dict[str, Dict[str, List[float | None]]] = {}
+    probe_scores: Dict[str, Dict[str, List[float | None]]] = {}
+    for arm, (arm_skill, arm_memory) in arms.items():
+        source_scores[arm] = {
+            source_id: _rollout_scores(
+                backend, source_by_id[source_id], arm_skill, arm_memory,
+                metric=metric, mixed_weight=mixed_weight, rollouts=rollouts,
+            )
+            for source_id in scored_source_ids
+        }
+        probe_scores[arm] = {
+            probe.id: _rollout_scores(
+                backend, probe, arm_skill, arm_memory,
+                metric=metric, mixed_weight=mixed_weight, rollouts=rollouts,
+            )
+            for probe in scored_probes
+        }
+
+    rows: List[Dict[str, Any]] = []
+    flagged = 0
+    invalid = 0
+    inconclusive = 0
+    gap_changes: List[float] = []
+    for probe in probes:
+        kind = next(
+            (tag.removeprefix("probe:") for tag in probe.tags if tag.startswith("probe:")),
+            "unknown",
+        )
+        if not distinct[probe.id]:
+            inconclusive += 1
+            rows.append({
+                "source_task_id": probe.derived_from,
+                "probe_task_id": probe.id,
+                "probe_kind": kind,
+                "baseline_source_score": None,
+                "baseline_probe_score": None,
+                "candidate_source_score": None,
+                "candidate_probe_score": None,
+                "baseline_gap": None,
+                "candidate_gap": None,
+                "gap_change": None,
+                "worsening_fraction": None,
+                "samples": {},
+                "distinct_samples": False,
+                "inconclusive_reason": "repeated_samples_unsupported",
+                "status": "inconclusive",
+            })
+            continue
+        samples = {
+            "baseline_source": source_scores["baseline"][probe.derived_from],
+            "baseline_probe": probe_scores["baseline"][probe.id],
+            "candidate_source": source_scores["candidate"][probe.derived_from],
+            "candidate_probe": probe_scores["candidate"][probe.id],
+        }
+        valid = all(
+            score is not None for scores in samples.values() for score in scores
+        )
+        if valid:
+            means = {name: _mean(scores) for name, scores in samples.items()}
+            baseline_gap = means["baseline_probe"] - means["baseline_source"]
+            candidate_gap = means["candidate_probe"] - means["candidate_source"]
+            gap_change = candidate_gap - baseline_gap
+            worsened = sum(
+                1
+                for index in range(rollouts)
+                if (
+                    samples["candidate_probe"][index]
+                    - samples["candidate_source"][index]
+                )
+                - (
+                    samples["baseline_probe"][index]
+                    - samples["baseline_source"][index]
+                )
+                < 0.0
+            )
+            worsening_fraction = worsened / rollouts
+            majority_worsened = worsened * 2 > rollouts
+            is_brittle = gap_change < -numeric_margin and majority_worsened
+            gap_changes.append(gap_change)
+        else:
+            means = {name: None for name in samples}
+            baseline_gap = candidate_gap = gap_change = None
+            worsening_fraction = None
+            is_brittle = True
+        if is_brittle:
+            flagged += 1
+        if not valid:
+            invalid += 1
+        rows.append({
+            "source_task_id": probe.derived_from,
+            "probe_task_id": probe.id,
+            "probe_kind": kind,
+            "baseline_source_score": means["baseline_source"],
+            "baseline_probe_score": means["baseline_probe"],
+            "candidate_source_score": means["candidate_source"],
+            "candidate_probe_score": means["candidate_probe"],
+            "baseline_gap": baseline_gap,
+            "candidate_gap": candidate_gap,
+            "gap_change": gap_change,
+            "worsening_fraction": worsening_fraction,
+            "samples": {name: list(scores) for name, scores in samples.items()},
+            "distinct_samples": True,
+            "status": "invalid" if not valid else ("brittle" if is_brittle else "stable"),
+        })
+
+    n = len(rows)
+    scored = n - inconclusive
+    return {
+        "enabled": True,
+        "factor": max(0, min(factor, MAX_PROBES_PER_TASK)),
+        "margin": numeric_margin,
+        "rollouts": rollouts,
+        "n_sources": len(source_ids),
+        "n_probes": n,
+        "n_flagged": flagged,
+        "n_invalid": invalid,
+        "n_inconclusive": inconclusive,
+        "brittleness_rate": (flagged / scored) if scored else 0.0,
+        "worst_gap_change": min(gap_changes) if gap_changes else None,
+        "conclusive": scored > 0,
+        "flagged": flagged > 0,
+        "rows": rows,
+    }
diff --git a/skillopt_sleep/backend.py b/skillopt_sleep/backend.py
index 99e41f418..5c1ea3eec 100644
--- a/skillopt_sleep/backend.py
+++ b/skillopt_sleep/backend.py
@@ -44,6 +44,61 @@ def skill_hash(content: str) -> str:
     return hashlib.sha256(content.encode("utf-8")).hexdigest()[:16]
 
 
+def accepts_sample_id(method: Any) -> bool:
+    """True when ``method`` can be called with a ``sample_id`` keyword.
+
+    Backends written before repeated rollouts existed may override
+    ``attempt`` or ``attempt_with_tools`` without that parameter. Callers use
+    this to keep calling such overrides safely and to report that their route
+    cannot produce distinct repeated samples.
+    """
+    import inspect
+
+    try:
+        params = inspect.signature(method).parameters.values()
+    except (TypeError, ValueError):
+        return False
+    return any(
+        param.kind is param.VAR_KEYWORD
+        or (
+            param.name == "sample_id"
+            and param.kind in (param.POSITIONAL_OR_KEYWORD, param.KEYWORD_ONLY)
+        )
+        for param in params
+    )
+
+
+def call_attempt_with_tools(
+    backend: "Backend",
+    task: TaskRecord,
+    skill: str,
+    memory: str,
+    tools: List[str],
+    *,
+    sample_id: int = 0,
+) -> Tuple[str, List[str]]:
+    """Run the tool route, forwarding ``sample_id`` when the backend accepts it.
+
+    Sample zero keeps the historical call shape, so existing overrides and
+    cached gate re-scoring are unchanged.
+    """
+    if sample_id and accepts_sample_id(backend.attempt_with_tools):
+        return backend.attempt_with_tools(
+            task, skill, memory, tools, sample_id=sample_id
+        )
+    return backend.attempt_with_tools(task, skill, memory, tools)
+
+
+def repeated_samples_distinct_for(backend: Any, *, tools: bool) -> bool:
+    """``backend.distinct_samples`` with a fallback for duck-typed backends
+    that predate it: only the ``sample_id`` signature can be inspected."""
+    probe = getattr(backend, "distinct_samples", None)
+    if callable(probe):
+        return bool(probe(tools=tools))
+    method = backend.attempt_with_tools if tools else backend.attempt
+    return accepts_sample_id(method)
+
+
 # ── Backend protocol ──────────────────────────────────────────────────────────
 
 class Backend:
@@ -61,21 +116,55 @@ def attempt(self, task: TaskRecord, skill: str, memory: str,
         raise NotImplementedError
 
     def attempt_with_tools(
-        self, task: TaskRecord, skill: str, memory: str, tools: List[str]
+        self,
+        task: TaskRecord,
+        skill: str,
+        memory: str,
+        tools: List[str],
+        sample_id: int = 0,
     ) -> Tuple[str, List[str]]:
         """Run the task while exposing real tools; return (response, tools_called).
 
         Default: no real tool loop — fall back to plain attempt and let the
         single-shot 'TOOL_CALL: <name>' marker convention surface intent. CLI
         backends override this to expose a genuinely callable tool.
+        ``sample_id`` is forwarded to ``attempt`` so repeated rollouts of a
+        tool task reach distinct cache entries instead of sample zero.
         """
-        resp = self.attempt(task, skill, memory)
+        resp = self._attempt_sample(task, skill, memory, sample_id)
         called: List[str] = []
         for t in tools:
             if re.search(r"(?i)\btool_call\s*:\s*%s\b" % re.escape(t), resp):
                 called.append(t)
         return resp, called
 
+    def _attempt_sample(
+        self, task: TaskRecord, skill: str, memory: str, sample_id: int
+    ) -> str:
+        if sample_id and accepts_sample_id(self.attempt):
+            return self.attempt(task, skill, memory, sample_id=sample_id)
+        return self.attempt(task, skill, memory)
+
+    def distinct_samples(self, *, tools: bool) -> bool:
+        """Whether repeated ``sample_id`` rollouts reach the model as distinct
+        samples on this route, rather than collapsing to one cached response.
+
+        * Text route: distinct when ``attempt`` accepts ``sample_id``; the
+          shipped attempt caches salt their keys with it.
+        * Tool route: ``attempt_with_tools`` must accept ``sample_id``. The
+          inherited marker fallback (and the mock tool model) answer through
+          ``attempt``, so they are distinct only when ``attempt`` is. The
+          shipped tool-loop overrides start a fresh, uncached agent process
+          on every call, so each call is already an independent sample.
+        """
+        if not tools:
+            return accepts_sample_id(self.attempt)
+        if not accepts_sample_id(self.attempt_with_tools):
+            return False
+        if type(self).attempt_with_tools in _ATTEMPT_DELEGATING_TOOL_ROUTES:
+            return accepts_sample_id(self.attempt)
+        return True
+
     def judge(self, task: TaskRecord, response: str) -> Tuple[float, float, str]:
         raise NotImplementedError
 
@@ -192,13 +281,13 @@ def attempt(self, task: TaskRecord, skill: str, memory: str,
             return f"approximately {mangled} (format not applied)"
         return "(attempted, no checkable reference)"
 
-    def attempt_with_tools(self, task, skill, memory, tools):
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
         # Deterministic tool model: the mock "calls" a tool iff the skill+memory
         # contains an explicit instruction to use it (a learned rule mentioning
         # the tool name or "search"). The deficient skill says NOT to, so
         # baseline calls nothing; a learned "use ./search" rule flips it.
         ctx = ((skill or "") + "\n" + (memory or "")).lower()
-        resp = self.attempt(task, skill, memory)
+        resp = self._attempt_sample(task, skill, memory, sample_id)
         called = []
         for t in (tools or []):
             tl = t.lower()
@@ -314,6 +403,14 @@ def _task_guardrail(pairs) -> str:
     )
 
 
+# Tool routes that answer through ``attempt`` rather than a real tool loop;
+# their sample independence is exactly that of ``attempt``.
+_ATTEMPT_DELEGATING_TOOL_ROUTES = frozenset({
+    Backend.attempt_with_tools,
+    MockBackend.attempt_with_tools,
+})
+
+
 class CliBackend(Backend):
     """Common logic for real CLI-driven backends (claude / codex).
 
@@ -797,7 +894,10 @@ def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
         self._detect_cli_error(out, proc.stderr or "")
         return out
 
-    def attempt_with_tools(self, task, skill, memory, tools):
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
+        # Each call starts a fresh, uncached agent process, so every call is
+        # already an independent sample; ``sample_id`` needs no salt here.
+        del sample_id
         # Expose a REAL, callable `search` tool (a shell shim that logs each
         # call) so the gbrain quick-answerer judge (tool_called=search) is
         # validated honestly: we detect the call from the shim's log, not from
@@ -1287,7 +1387,11 @@ def attempt_with_tools(
         skill: str,
         memory: str,
         tools: List[str],
+        sample_id: int = 0,
     ) -> Tuple[str, List[str]]:
+        # Each call starts a fresh, uncached agent process, so every call is
+        # already an independent sample; ``sample_id`` needs no salt here.
+        del sample_id
         self.last_call_error = ""
         if not self.tool_replay:
             self.last_call_error = (
@@ -1544,7 +1648,10 @@ def _call(self, prompt: str, *, max_tokens: int = 1024, retries: int = 3) -> str
                 _t.sleep(min(6.0, (2 ** attempt) * 0.5) + _r.random() * 0.3)
         return out
 
-    def attempt_with_tools(self, task, skill, memory, tools):
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
+        # Each call starts a fresh, uncached agent process, so every call is
+        # already an independent sample; ``sample_id`` needs no salt here.
+        del sample_id
         # Codex exec runs in a sandbox with shell access; expose the same real
         # `search` shim and let it run (workspace-write so the shim can log).
         import tempfile, shutil, stat
@@ -1808,7 +1915,10 @@ def _parse_jsonl_response(raw: str) -> str:
                     parts.append(content)
         return "\n".join(parts).strip()
 
-    def attempt_with_tools(self, task, skill, memory, tools):
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
+        # Each call starts a fresh, uncached agent process, so every call is
+        # already an independent sample; ``sample_id`` needs no salt here.
+        del sample_id
         # Expose REAL, callable tool shims in the working directory so the
         # gbrain quick-answerer judge (tool_called=search) is validated
         # honestly: we detect each call from the shim's log, not from a
@@ -2143,8 +2253,9 @@ def attempt_with_tools(
         skill: str,
         memory: str,
         tools: List[str],
+        sample_id: int = 0,
     ) -> Tuple[str, List[str]]:
-        del task, skill, memory, tools
+        del task, skill, memory, tools, sample_id
         raise self._error(
             "Cursor tool-aware replay is temporarily disabled pending live "
             "Cursor permission-boundary validation"
@@ -2173,8 +2284,14 @@ def __init__(self, target: Backend, optimizer: Backend) -> None:
     def attempt(self, task, skill, memory, sample_id: int = 0):
         return self.target.attempt(task, skill, memory, sample_id=sample_id)
 
-    def attempt_with_tools(self, task, skill, memory, tools):
-        return self.target.attempt_with_tools(task, skill, memory, tools)
+    def attempt_with_tools(self, task, skill, memory, tools, sample_id: int = 0):
+        return call_attempt_with_tools(
+            self.target, task, skill, memory, tools, sample_id=sample_id
+        )
+
+    def distinct_samples(self, *, tools: bool) -> bool:
+        # Attempts run on the target, so its routes decide sample independence.
+        return repeated_samples_distinct_for(self.target, tools=tools)
 
     def judge(self, task, response):
         # local rule/exact judging needs no model; delegate to target which
diff --git a/skillopt_sleep/config.py b/skillopt_sleep/config.py
index d4a008d1f..356dfd024 100644
--- a/skillopt_sleep/config.py
+++ b/skillopt_sleep/config.py
@@ -70,6 +70,10 @@
     # ── 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 worsening of the candidate-vs-baseline gap in [0, 1]
+    "dream_adversarial_rollouts": 1,      # samples per task/arm; blocking requires >= 2
     "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 5b80b4a98..ba3d20dbd 100644
--- a/skillopt_sleep/consolidate.py
+++ b/skillopt_sleep/consolidate.py
@@ -12,6 +12,11 @@
 from dataclasses import dataclass, field
 from typing import List, Tuple
 
+from skillopt_sleep.adversarial import (
+    MAX_PROBE_ROLLOUTS,
+    MIN_BLOCKING_ROLLOUTS,
+    evaluate_adversarial_probes,
+)
 from skillopt_sleep.backend import Backend
 
 # Self-contained validation gate (vendored from SkillOpt; zero dependency on the
@@ -187,6 +192,10 @@ 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,
+    dream_adversarial_rollouts: int = 1,
     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 +211,46 @@ 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]"
+            )
+        if isinstance(dream_adversarial_rollouts, bool) or not isinstance(
+            dream_adversarial_rollouts, int
+        ):
+            raise ValueError(
+                "dream_adversarial_rollouts must be an integer in "
+                f"[1, {MAX_PROBE_ROLLOUTS}]"
+            )
+        if not 1 <= dream_adversarial_rollouts <= MAX_PROBE_ROLLOUTS:
+            raise ValueError(
+                "dream_adversarial_rollouts must be an integer in "
+                f"[1, {MAX_PROBE_ROLLOUTS}]"
+            )
+        if (
+            dream_adversarial_blocking
+            and dream_adversarial_rollouts < MIN_BLOCKING_ROLLOUTS
+        ):
+            raise ValueError(
+                "dream_adversarial_blocking requires "
+                f"dream_adversarial_rollouts >= {MIN_BLOCKING_ROLLOUTS} so a "
+                "single stochastic sample can never reject a candidate; keep "
+                "the probes advisory for single-rollout runs"
+            )
     from skillopt_sleep import evidence as evlog
     ev = evlog.get(backend)
     train_tasks, val_tasks, holdout_leaked = _split(tasks)
@@ -261,16 +310,75 @@ 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,
+                baseline_skill=cand_skill,
+                baseline_memory=cand_memory,
+                factor=dream_adversarial,
+                metric=gate_metric,
+                mixed_weight=gate_mixed_weight,
+                margin=dream_adversarial_margin,
+                rollouts=dream_adversarial_rollouts,
+            )
+            blocked = bool(
+                dream_adversarial_blocking
+                and (probe["flagged"] or not probe["conclusive"])
+            )
+            probe["blocking"] = bool(dream_adversarial_blocking)
+            probe["blocked"] = blocked
+            if not blocked:
+                probe["block_reason"] = ""
+            elif probe["conclusive"]:
+                probe["block_reason"] = "brittle_score_drop"
+            elif probe["n_probes"]:
+                probe["block_reason"] = "inconclusive_repeated_samples_unsupported"
+            else:
+                probe["block_reason"] = "inconclusive_no_probes"
+            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 +391,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 f1d7bf027..25c08c0cf 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,19 @@ 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))}, "
+            f"rollouts={int(cfg.get('dream_adversarial_rollouts', 1) or 1)})",
+        )
     gate_on = str(cfg.get("gate_mode", "on")).strip().lower() not in {
         "off", "none", "false", "greedy",
     }
@@ -341,9 +354,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 +384,53 @@ 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}, baseline-relative, "
+                    f"rollouts={int(probe.get('rollouts', 1) or 1)}): "
+                    f"{int(probe.get('n_flagged', 0) or 0)} flagged / "
+                    f"{int(probe.get('n_probes', 0) or 0)} total; "
+                    f"worst gap change {_report_score(probe.get('worst_gap_change'))}."
+                )
+                n_inconclusive = int(probe.get("n_inconclusive", 0) or 0)
+                if n_inconclusive:
+                    lines.append(
+                        f"{n_inconclusive} probe(s) inconclusive: their replay route "
+                        "cannot produce distinct repeated samples, so they were not "
+                        "scored and cannot flag the candidate."
+                    )
+                if not probe.get("conclusive", False):
+                    lines.append(
+                        "No conclusive probe was scored; blocking mode fails "
+                        "closed for this candidate."
+                        if int(probe.get("n_probes", 0) or 0)
+                        else "No eligible probe was generated; blocking mode fails "
+                        "closed for this candidate."
+                    )
+                lines.append("")
+                lines.append(
+                    "| Source | Probe | Variant | Baseline source | Baseline probe | "
+                    "Candidate source | Candidate probe | Gap change | 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", ""))
+                    baseline_source = _report_score(row.get("baseline_source_score"))
+                    baseline_probe = _report_score(row.get("baseline_probe_score"))
+                    candidate_source = _report_score(row.get("candidate_source_score"))
+                    candidate_probe = _report_score(row.get("candidate_probe_score"))
+                    gap_change = _report_score(row.get("gap_change"))
+                    status = _markdown_table_text(row.get("status", ""))
+                    lines.append(
+                        f"| `{source_id}` | `{probe_id}` | {kind} | "
+                        f"{baseline_source} | {baseline_probe} | "
+                        f"{candidate_source} | {candidate_probe} | {gap_change} | {status} |"
+                    )
+                lines.append("")
     if report.edits:
         lines.append("## Accepted edits")
         for e in report.edits:
@@ -698,6 +763,19 @@ 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
+                ),
+                "dream_adversarial_rollouts": cfg.get(
+                    "dream_adversarial_rollouts", 1
+                ),
+            })
         cycle_config["opencode_tool_replay"] = (
             cfg.get("opencode_tool_replay", False) is True
         )
@@ -844,8 +922,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 +938,12 @@ 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),
+            dream_adversarial_rollouts=cfg.get("dream_adversarial_rollouts", 1),
             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 +1038,16 @@ 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
+                    ),
+                    dream_adversarial_rollouts=cfg.get(
+                        "dream_adversarial_rollouts", 1
+                    ),
                     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 +1134,18 @@ 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
+                            ),
+                            "dream_adversarial_rollouts": cfg.get(
+                                "dream_adversarial_rollouts", 1
+                            ),
+                        } 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 9906e07d5..090169297 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,10 @@ 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,
+    dream_adversarial_rollouts: int = 1,
     gate_mode: str = "on",
     evolve_skill: bool = True,
     evolve_memory: bool = True,
@@ -140,8 +147,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 +170,12 @@ 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,
+        dream_adversarial_rollouts=dream_adversarial_rollouts,
+        gate_mode=gate_mode,
         rollouts_k=dream_rollouts, evolve_skill=evolve_skill,
         evolve_memory=evolve_memory, night=night,
     )
diff --git a/skillopt_sleep/replay.py b/skillopt_sleep/replay.py
index 1502a71cf..20c48bcaa 100644
--- a/skillopt_sleep/replay.py
+++ b/skillopt_sleep/replay.py
@@ -12,7 +12,11 @@
 
 from typing import List, Tuple
 
-from skillopt_sleep.backend import Backend
+from skillopt_sleep.backend import (
+    Backend,
+    call_attempt_with_tools,
+    repeated_samples_distinct_for,
+)
 from skillopt_sleep.types import ReplayResult, TaskRecord
 
 
@@ -27,6 +31,12 @@ def _required_tools(task: TaskRecord) -> List[str]:
     return tools
 
 
+def repeated_samples_distinct(backend: Backend, task: TaskRecord) -> bool:
+    """Whether ``replay_one`` obtains distinct samples for different
+    ``sample_id`` values on the route this task takes (tool or text)."""
+    return repeated_samples_distinct_for(backend, tools=bool(_required_tools(task)))
+
+
 def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
                sample_id: int = 0) -> ReplayResult:
     """``sample_id`` distinguishes repeated dream rollouts of the same
@@ -38,7 +48,9 @@ def replay_one(backend: Backend, task: TaskRecord, skill: str, memory: str,
     t0 = time.time()
     tok_before = backend.tokens_used()
     if tools:
-        response, tools_called = backend.attempt_with_tools(task, skill, memory, tools)
+        response, tools_called = call_attempt_with_tools(
+            backend, task, skill, memory, tools, sample_id=sample_id
+        )
     else:
         response = backend.attempt(task, skill, memory, sample_id=sample_id)
     latency_ms = (time.time() - t0) * 1000.0
diff --git a/tests/test_adversarial_dream.py b/tests/test_adversarial_dream.py
new file mode 100644
index 000000000..d3ed9f567
--- /dev/null
+++ b/tests/test_adversarial_dream.py
@@ -0,0 +1,1121 @@
+"""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,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+        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_gap_change"] == -1.0
+    assert {row["status"] for row in report["rows"]} == {"brittle"}
+    for row in report["rows"]:
+        assert row["baseline_source_score"] == 0.0
+        assert row["baseline_probe_score"] == 0.0
+        assert row["candidate_source_score"] == 1.0
+        assert row["candidate_probe_score"] == 0.0
+        assert row["gap_change"] == -1.0
+
+
+def test_probe_report_keeps_a_surface_robust_rule_stable() -> None:
+    report = evaluate_adversarial_probes(
+        _RobustCandidateBackend(),
+        [_task("train")],
+        _CandidateBackend.RULE,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+        factor=3,
+    )
+
+    assert report["n_flagged"] == 0
+    assert report["flagged"] is False
+    assert report["worst_gap_change"] == 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,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+    )
+
+    assert report["flagged"] is False
+    # source + one probe, each scored under the baseline AND candidate arms
+    assert target.attempt_calls == 4
+    assert target.judge_calls == 4
+    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,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+    )
+
+    assert report["flagged"] is True
+    assert report["n_invalid"] == 1
+    assert report["rows"][0]["candidate_probe_score"] is None
+    assert report["rows"][0]["gap_change"] 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,
+            "",
+        baseline_skill="",
+        baseline_memory="",
+            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",
+        ),
+        (
+            {"dream_adversarial": 1, "dream_adversarial_rollouts": True},
+            "rollouts must be an integer",
+        ),
+        (
+            {"dream_adversarial": 1, "dream_adversarial_rollouts": 0},
+            "rollouts must be an integer",
+        ),
+        (
+            {"dream_adversarial": 1, "dream_adversarial_rollouts": 99},
+            "rollouts must be an integer",
+        ),
+        (
+            {
+                "dream_adversarial": 1,
+                "dream_adversarial_blocking": True,
+                "dream_adversarial_rollouts": 1,
+            },
+            "blocking requires",
+        ),
+    ],
+)
+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,
+        dream_adversarial_rollouts=2,
+    )
+
+    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,
+        dream_adversarial_rollouts=2,
+    )
+
+    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,
+        dream_adversarial_rollouts=2,
+        auto_adopt=False,
+    )
+
+    tasks = _candidate_tasks()
+    tasks[0].id = "train|<script>\n# heading"
+    outcome = run_sleep_cycle(
+        config,
+        seed_tasks=tasks,
+        backend=_CandidateBackend(),
+    )
+
+    with open(
+        os.path.join(outcome.staging_dir, "diagnostics.json"),
+        encoding="utf-8",
+    ) as handle:
+        diagnostics = json.load(handle)
+    with open(
+        os.path.join(outcome.staging_dir, "report.md"),
+        encoding="utf-8",
+    ) as handle:
+        markdown = handle.read()
+    assert diagnostics["dream_adversarial"] == 2
+    assert diagnostics["dream_adversarial_blocking"] is False
+    assert diagnostics["dream_adversarial_margin"] == 0.015
+    assert diagnostics["dream_adversarial_rollouts"] == 2
+    assert diagnostics["gate_trials"] == outcome.report.gate_trials
+    assert diagnostics["gate_trials"][0]["adversarial_probe"]["n_flagged"] == 2
+    assert "adversarial dream probes: advisory" in markdown
+    assert (
+        "Adversarial probes (advisory, baseline-relative, rollouts=2): "
+        "2 flagged / 2 total" in markdown
+    )
+    assert "<script>" not in markdown
+    assert "&lt;script&gt;" in markdown
+
+
+def test_multi_skill_cycle_keeps_blocking_probe_rollouts(tmp_path) -> None:
+    project = tmp_path / "project"
+    project.mkdir()
+    claude_home = tmp_path / ".claude"
+    skill_dir = claude_home / "skills" / "demo-skill"
+    skill_dir.mkdir(parents=True)
+    (skill_dir / "SKILL.md").write_text("# demo skill\n", encoding="utf-8")
+    config = load_config(
+        invoked_project=str(project),
+        projects="invoked",
+        backend="mock",
+        state_dir=str(tmp_path / "state"),
+        claude_home=str(claude_home),
+        managed_skill_name="skillopt-sleep-learned",
+        multi_skill_fanout=True,
+        evolve_memory=False,
+        dream_adversarial=1,
+        dream_adversarial_blocking=True,
+        dream_adversarial_rollouts=2,
+        auto_adopt=False,
+    )
+
+    outcome = run_sleep_cycle(
+        config,
+        seed_tasks=_candidate_tasks(),
+        backend=_RobustCandidateBackend(),
+    )
+
+    assert len(outcome.report.skill_groups) == 1
+    group = outcome.report.skill_groups[0]
+    assert group.skill_name == "demo-skill"
+    assert group.status == "consolidated"
+    assert group.accepted is True
+
+
+@pytest.mark.parametrize("blocking", [False, True], ids=["advisory", "blocking"])
+@pytest.mark.parametrize(
+    ("backend_type", "probe_score", "gap_change", "status"),
+    [
+        (_CandidateBackend, 0.0, -1.0, "brittle"),
+        (_RobustCandidateBackend, 1.0, 0.0, "stable"),
+    ],
+    ids=["brittle", "robust"],
+)
+def test_cycle_report_renders_baseline_relative_probe_scores(
+    tmp_path, blocking, backend_type, probe_score, gap_change, status
+) -> 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=1,
+        dream_adversarial_blocking=blocking,
+        dream_adversarial_rollouts=2,
+        auto_adopt=False,
+    )
+
+    outcome = run_sleep_cycle(
+        config, seed_tasks=_candidate_tasks(), backend=backend_type()
+    )
+
+    row = outcome.report.gate_trials[0]["adversarial_probe"]["rows"][0]
+    assert row["baseline_source_score"] == 0.0
+    assert row["baseline_probe_score"] == 0.0
+    assert row["candidate_source_score"] == 1.0
+    assert row["candidate_probe_score"] == probe_score
+    assert row["gap_change"] == gap_change
+    assert row["status"] == status
+    with open(
+        os.path.join(outcome.staging_dir, "report.md"), encoding="utf-8"
+    ) as handle:
+        markdown = handle.read()
+    assert (
+        "| Source | Probe | Variant | Baseline source | Baseline probe | "
+        "Candidate source | Candidate probe | Gap change | Status |"
+    ) in markdown
+    assert (
+        "| `train` | `train_adversarial_request-frame` | request-frame | "
+        f"0.000 | 0.000 | 1.000 | {probe_score:.3f} | {gap_change:.3f} | {status} |"
+    ) in markdown
+
+
+class _EquallySensitiveBackend(_CandidateBackend):
+    """Frame sensitivity exists identically with and without the candidate."""
+
+    def attempt(self, task, skill, memory, sample_id=0):
+        return task.reference if task.intent in self.HARVESTED_INTENTS else "wrong"
+
+
+class _ImprovesBothKeepsGapBackend(MockBackend):
+    """The candidate improves source AND probe but keeps the pre-existing gap."""
+
+    SCORES = {
+        (False, False): 0.4,   # baseline arm, source task
+        (False, True): 0.1,    # baseline arm, probe task
+        (True, False): 0.9,    # candidate arm, source task
+        (True, True): 0.6,     # candidate arm, probe task
+    }
+
+    def attempt(self, task, skill, memory, sample_id=0):
+        has_rule = _CandidateBackend.RULE in f"{skill}\n{memory}"
+        return f"marker:{int(has_rule)}:{int(task.origin == 'dream')}"
+
+    def judge(self, task, response):
+        _, has_rule, is_probe = response.split(":")
+        value = self.SCORES[(has_rule == "1", is_probe == "1")]
+        return value, value, "graded fixture"
+
+
+class _SingleFluctuationBackend(MockBackend):
+    """One probe rollout dips under the candidate; the repeat does not."""
+
+    def attempt(self, task, skill, memory, sample_id=0):
+        has_rule = _CandidateBackend.RULE in f"{skill}\n{memory}"
+        dips = has_rule and task.origin == "dream" and sample_id == 0
+        return f"marker:{int(dips)}"
+
+    def judge(self, task, response):
+        value = 0.8 if response.endswith(":1") else 1.0
+        return value, value, "graded fixture"
+
+
+def test_equal_frame_sensitivity_in_baseline_and_candidate_is_not_flagged() -> None:
+    report = evaluate_adversarial_probes(
+        _EquallySensitiveBackend(),
+        [_task("train")],
+        _CandidateBackend.RULE,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+        factor=2,
+        rollouts=2,
+    )
+
+    assert report["n_flagged"] == 0
+    assert report["flagged"] is False
+    for row in report["rows"]:
+        assert row["baseline_gap"] == row["candidate_gap"]
+        assert row["gap_change"] == 0.0
+        assert row["status"] == "stable"
+
+
+def test_candidate_that_improves_both_but_keeps_the_gap_is_not_flagged() -> None:
+    report = evaluate_adversarial_probes(
+        _ImprovesBothKeepsGapBackend(),
+        [_task("train")],
+        _CandidateBackend.RULE,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+        factor=2,
+        rollouts=2,
+    )
+
+    assert report["flagged"] is False
+    for row in report["rows"]:
+        assert row["candidate_source_score"] > row["baseline_source_score"]
+        assert row["candidate_probe_score"] > row["baseline_probe_score"]
+        assert row["gap_change"] == 0.0
+        assert row["status"] == "stable"
+
+
+def test_single_rollout_fluctuation_never_flags_under_majority_rule() -> None:
+    report = evaluate_adversarial_probes(
+        _SingleFluctuationBackend(),
+        [_task("train")],
+        _CandidateBackend.RULE,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+        factor=1,
+        rollouts=2,
+    )
+
+    row = report["rows"][0]
+    assert row["gap_change"] < 0.0
+    assert row["worsening_fraction"] == 0.5
+    assert row["status"] == "stable"
+    assert report["flagged"] is False
+
+
+def test_evidence_rows_retain_all_four_scores_and_samples() -> None:
+    report = evaluate_adversarial_probes(
+        _CandidateBackend(),
+        [_task("train")],
+        _CandidateBackend.RULE,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+        factor=1,
+        rollouts=2,
+    )
+
+    row = report["rows"][0]
+    for key in (
+        "baseline_source_score",
+        "baseline_probe_score",
+        "candidate_source_score",
+        "candidate_probe_score",
+        "baseline_gap",
+        "candidate_gap",
+        "gap_change",
+        "worsening_fraction",
+    ):
+        assert key in row
+    for arm in (
+        "baseline_source",
+        "baseline_probe",
+        "candidate_source",
+        "candidate_probe",
+    ):
+        assert len(row["samples"][arm]) == 2
+    json.dumps(report, allow_nan=False)
+
+
+@pytest.mark.parametrize("rollouts", [True, 0, -1, 9, 1.5, "2"])
+def test_probe_rollouts_rejects_out_of_range_or_ambiguous_values(rollouts) -> None:
+    with pytest.raises(ValueError, match="rollouts must be an integer"):
+        evaluate_adversarial_probes(
+            _RobustCandidateBackend(),
+            [_task("train")],
+            _CandidateBackend.RULE,
+            "",
+        baseline_skill="",
+        baseline_memory="",
+            rollouts=rollouts,
+        )
+
+
+@pytest.mark.parametrize(
+    "intent",
+    [
+        "Can you swim?",
+        "Could you ever forgive me?",
+        "Would you like some tea?",
+        "Can you believe the score?",
+        "I want you to know this matters.",
+        "I need you to be honest with me.",
+    ],
+)
+def test_polite_frame_never_reframes_ability_permission_or_desire(intent) -> None:
+    from skillopt_sleep.adversarial import _strip_polite_frame, _variant_intents
+
+    assert _strip_polite_frame(intent) == ""
+    assert all(kind != "request-frame" for kind, _ in _variant_intents(intent))
+
+
+@pytest.mark.parametrize(
+    ("intent", "expected"),
+    [
+        ("Please add validation to the signup page", "Add validation to the signup page"),
+        ("please return ok", "Return ok"),
+        ("Could you please add input validation?", "Add input validation."),
+        ("can you please rerun the failing job", "Rerun the failing job"),
+        ("Please can you swim?", "Can you swim?"),
+    ],
+)
+def test_polite_frame_reframes_only_politeness_marked_requests(intent, expected) -> None:
+    from skillopt_sleep.adversarial import _strip_polite_frame
+
+    assert _strip_polite_frame(intent) == expected
+
+
+# ── Repeated-sample independence on tool and text routes ─────────────────────
+
+_PI_RULE = "Always call the search tool and finish with VERIFIED."
+_PI_INTENTS = (
+    "Please look up the current release notes before answering",
+    "Please check the changelog for the latest breaking change",
+    "Please find the deprecation date of the old endpoint",
+    "Please confirm which version introduced the retry flag",
+)
+_PI_CHECKS = {
+    "tool": {"op": "tool_called", "arg": "search"},
+    "text": {"op": "contains", "arg": "VERIFIED"},
+}
+
+
+class _FakePiProvider:
+    """Stands in for the Pi child process, the only faked boundary.
+
+    The miner, reflection, replay, cache, gate, probes, and report all run the
+    shipped code. Each call is recorded so tests can count what actually
+    reached the provider. The first candidate-probe call of the probe phase
+    fails once; every other candidate call succeeds and baseline calls fail.
+    """
+
+    GOOD = "TOOL_CALL: search\nVERIFIED"
+    BAD = "I answered from memory."
+
+    def __init__(self, backend, check) -> None:
+        self.backend = backend
+        self.check = check
+        self.probe_calls: list[tuple[str, str]] = []
+        self.failed_once = False
+
+    def __call__(self, cmd, *, input, **kwargs):
+        import subprocess
+
+        prompt = input
+        if prompt.startswith("You are mining"):
+            intent = next(text for text in _PI_INTENTS if text in prompt)
+            out = json.dumps(
+                [{"intent": intent, "checks": [self.check], "satisfied": False}]
+            )
+        elif prompt.startswith("You are SkillOpt's optimizer"):
+            current = prompt.split("# Recurring failures", 1)[0]
+            out = "[]" if _PI_RULE in current else json.dumps(
+                [{"op": "add", "content": _PI_RULE, "rationale": "fixture"}]
+            )
+        elif prompt.startswith("Complete the following task"):
+            candidate = _PI_RULE in prompt
+            out = self.GOOD if candidate else self.BAD
+            if self.backend.evidence_phase.startswith("adversarial_probe"):
+                arm = "candidate" if candidate else "baseline"
+                role = (
+                    "source" if any(text in prompt for text in _PI_INTENTS) else "probe"
+                )
+                self.probe_calls.append((arm, role))
+                if candidate and role == "probe" and not self.failed_once:
+                    self.failed_once = True
+                    out = self.BAD
+        else:
+            out = "{}"
+        return subprocess.CompletedProcess(cmd, 0, stdout=out, stderr="")
+
+
+@pytest.mark.parametrize("route", ["tool", "text"])
+def test_miner_derived_pi_cycle_obtains_distinct_probe_rollouts(tmp_path, route) -> None:
+    """A miner-produced task replayed through the shipped Pi backend must give
+    each probe rollout its own provider sample on both routes.
+
+    Before the fix, the inherited tool fallback dropped ``sample_id`` and served
+    every rollout from the sample-zero cache entry: one failed candidate-probe
+    response counted as three failures and blocked a candidate that the text
+    route (the positive control) accepts.
+    """
+    from collections import Counter
+    from unittest import mock
+
+    from skillopt_sleep.backend import PiCliBackend
+    from skillopt_sleep.llm_miner import make_llm_miner
+    from skillopt_sleep.mine import mine
+    from skillopt_sleep.types import SessionDigest
+
+    rollouts = 3
+    backend = PiCliBackend()
+    provider = _FakePiProvider(backend, _PI_CHECKS[route])
+    digests = [
+        SessionDigest(
+            session_id=f"session-{index}",
+            project="/project",
+            user_prompts=[intent],
+            assistant_finals=["done"],
+        )
+        for index, intent in enumerate(_PI_INTENTS)
+    ]
+    project = tmp_path / "project"
+    project.mkdir()
+    config = load_config(
+        invoked_project=str(project),
+        projects="invoked",
+        backend="pi",
+        state_dir=str(tmp_path / "state"),
+        claude_home=str(tmp_path / ".claude"),
+        evolve_memory=False,
+        dream_adversarial=1,
+        dream_adversarial_blocking=True,
+        dream_adversarial_rollouts=rollouts,
+        auto_adopt=False,
+    )
+    with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=provider):
+        tasks = mine(digests, llm_miner=make_llm_miner(backend), max_tasks=10, seed=42)
+        outcome = run_sleep_cycle(config, seed_tasks=tasks, backend=backend)
+
+    assert {task.judge["checks"][0]["op"] for task in tasks} == {
+        _PI_CHECKS[route]["op"]
+    }
+    trial = outcome.report.gate_trials[0]
+    probe = trial["adversarial_probe"]
+    assert trial["accepted"] is True
+    assert trial["blocked_by_adversarial"] is False
+    assert probe["n_flagged"] == 0
+    n_pairs = probe["n_probes"]
+    assert n_pairs == probe["n_sources"] == 3
+    # The single provider failure is visible as exactly one failed rollout.
+    candidate_probe_samples = [
+        score for row in probe["rows"] for score in row["samples"]["candidate_probe"]
+    ]
+    assert candidate_probe_samples.count(0.0) == 1
+    assert provider.failed_once
+
+    # Provider-boundary call counts in the probe phase. Probe prompts are new,
+    # so every rollout of every probe reaches the provider in both arms. The
+    # candidate's source prompts are also new. Baseline source rollout zero is
+    # byte-identical to the training replay under the same documents, so it is
+    # served from that cache entry; rollouts one and two still reach Pi.
+    calls = Counter(provider.probe_calls)
+    assert calls[("candidate", "probe")] == rollouts * n_pairs
+    assert calls[("baseline", "probe")] == rollouts * n_pairs
+    assert calls[("candidate", "source")] == rollouts * n_pairs
+    assert calls[("baseline", "source")] == (rollouts - 1) * n_pairs
+
+
+def test_inherited_tool_fallback_forwards_sample_id_to_the_attempt_cache() -> None:
+    from unittest import mock
+
+    from skillopt_sleep.backend import PiCliBackend
+    from skillopt_sleep.replay import replay_one
+
+    backend = PiCliBackend()
+    provider = _FakePiProvider(backend, _PI_CHECKS["tool"])
+    task = TaskRecord(
+        id="tool-task",
+        project="/project",
+        intent=_PI_INTENTS[0],
+        reference_kind="rule",
+        judge={"kind": "rule", "checks": [_PI_CHECKS["tool"]]},
+    )
+    with mock.patch(
+        "skillopt_sleep.backend.subprocess.run", side_effect=provider
+    ) as run:
+        for sample_id in (0, 1, 2, 0, 1):
+            result = replay_one(backend, task, _PI_RULE, "", sample_id=sample_id)
+            assert result.tools_called == ["search"]
+            assert result.hard == 1.0
+    # Three distinct samples reach Pi; repeats of a sample id stay cached.
+    assert run.call_count == 3
+    assert backend.distinct_samples(tools=True) is True
+    assert backend.distinct_samples(tools=False) is True
+
+
+@pytest.mark.parametrize("route", ["tool", "text"])
+def test_dream_rollouts_of_a_tool_task_reach_the_provider_as_distinct_samples(
+    route,
+) -> None:
+    """The same dropped ``sample_id`` also collapsed mainline dream rollouts:
+    ``multi_rollout`` of a ``tool_called`` task made one Pi call and reported
+    K identical scores, so contrastive reflection never saw any spread."""
+    import itertools
+    import subprocess
+    from unittest import mock
+
+    from skillopt_sleep.backend import PiCliBackend
+    from skillopt_sleep.rollout import multi_rollout
+
+    counter = itertools.count()
+
+    def stochastic_provider(cmd, *, input, **kwargs):
+        # Alternates between calling the tool and answering without it.
+        if next(counter) % 2 == 0:
+            out = "TOOL_CALL: search\nanswer"
+        else:
+            out = "answer without the tool"
+        return subprocess.CompletedProcess(cmd, 0, stdout=out, stderr="")
+
+    check = (
+        {"op": "tool_called", "arg": "search"}
+        if route == "tool"
+        else {"op": "contains", "arg": "TOOL_CALL"}
+    )
+    task = TaskRecord(
+        id="dream-task",
+        project="/project",
+        intent="Please look it up",
+        reference_kind="rule",
+        judge={"kind": "rule", "checks": [check]},
+    )
+    with mock.patch(
+        "skillopt_sleep.backend.subprocess.run", side_effect=stochastic_provider
+    ) as run:
+        rollout_set = multi_rollout(PiCliBackend(), task, "skill", "", k=4, workers=1)
+    assert run.call_count == 4
+    assert [attempt.hard for attempt in rollout_set.attempts] == [1.0, 0.0, 1.0, 0.0]
+
+
+def test_dual_backend_forwards_tool_sample_ids_to_the_target() -> None:
+    seen: list[int] = []
+
+    class _Target(MockBackend):
+        def attempt(self, task, skill, memory, sample_id=0):
+            seen.append(sample_id)
+            return "TOOL_CALL: search"
+
+    task = TaskRecord(
+        id="tool-task",
+        project="/project",
+        intent="Please search",
+        reference_kind="rule",
+        judge={"kind": "rule", "checks": [{"op": "tool_called", "arg": "search"}]},
+    )
+    from skillopt_sleep.replay import repeated_samples_distinct, replay_one
+
+    dual = DualBackend(_Target(), MockBackend())
+    for sample_id in range(3):
+        replay_one(dual, task, "", "", sample_id=sample_id)
+    assert seen == [0, 1, 2]
+    assert repeated_samples_distinct(dual, task) is True
+
+
+def test_every_shipped_backend_route_claims_distinct_samples() -> None:
+    from skillopt_sleep import backend as backend_module
+
+    shipped = [
+        backend_module.MockBackend,
+        backend_module.PiCliBackend,
+        backend_module.ClaudeCliBackend,
+        backend_module.OpenCodeCliBackend,
+        backend_module.CodexCliBackend,
+        backend_module.CopilotCliBackend,
+        backend_module.CursorCliBackend,
+        backend_module.AzureOpenAIBackend,
+        backend_module.AzureResponsesBackend,
+    ]
+    for cls in shipped:
+        # distinct_samples inspects method signatures only, so no CLI, network,
+        # or credential set-up is needed to evaluate the route contract.
+        instance = cls.__new__(cls)
+        assert instance.distinct_samples(tools=False) is True, cls.__name__
+        assert instance.distinct_samples(tools=True) is True, cls.__name__
+
+
+class _LegacyToolBackend(_RobustCandidateBackend):
+    """A third-party backend written before repeated rollouts existed."""
+
+    def __init__(self) -> None:
+        self.attempt_calls = 0
+
+    def attempt(self, task, skill, memory):  # no sample_id parameter
+        self.attempt_calls += 1
+        if self.RULE not in f"{skill}\n{memory}":
+            return "wrong"
+        return task.reference
+
+    def attempt_with_tools(self, task, skill, memory, tools):
+        return self.attempt(task, skill, memory), list(tools)
+
+
+def _tool_tasks() -> list[TaskRecord]:
+    tasks = _candidate_tasks()
+    for task in tasks:
+        task.reference_kind = "rule"
+        task.judge = {
+            "kind": "rule",
+            "checks": [
+                {"op": "tool_called", "arg": "search"},
+                {"op": "contains", "arg": "ok"},
+            ],
+        }
+    return tasks
+
+
+def test_route_without_distinct_samples_is_inconclusive_not_brittle() -> None:
+    backend = _LegacyToolBackend()
+    before = backend.attempt_calls
+    report = evaluate_adversarial_probes(
+        backend,
+        _tool_tasks(),
+        _LegacyToolBackend.RULE,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+        rollouts=3,
+    )
+    assert backend.distinct_samples(tools=True) is False
+    assert report["n_probes"] == 1
+    assert report["n_inconclusive"] == 1
+    assert report["n_flagged"] == 0
+    assert report["flagged"] is False
+    assert report["conclusive"] is False
+    assert report["brittleness_rate"] == 0.0
+    row = report["rows"][0]
+    assert row["status"] == "inconclusive"
+    assert row["distinct_samples"] is False
+    assert row["inconclusive_reason"] == "repeated_samples_unsupported"
+    assert row["gap_change"] is None and row["samples"] == {}
+    # Collapsed rollouts are not replayed at all, so they cost nothing.
+    assert backend.attempt_calls == before
+    json.dumps(report, allow_nan=False)
+
+
+def test_single_rollout_makes_no_independence_claim_on_a_legacy_route() -> None:
+    report = evaluate_adversarial_probes(
+        _LegacyToolBackend(),
+        _tool_tasks(),
+        _LegacyToolBackend.RULE,
+        "",
+        baseline_skill="",
+        baseline_memory="",
+        rollouts=1,
+    )
+    assert report["n_inconclusive"] == 0
+    assert report["conclusive"] is True
+    assert report["rows"][0]["status"] == "stable"
+
+
+@pytest.mark.parametrize("blocking", [False, True], ids=["advisory", "blocking"])
+def test_legacy_route_inconclusive_evidence_in_the_gate_and_report(
+    tmp_path, blocking
+) -> 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=1,
+        dream_adversarial_blocking=blocking,
+        dream_adversarial_rollouts=2,
+        auto_adopt=False,
+    )
+    outcome = run_sleep_cycle(
+        config, seed_tasks=_tool_tasks(), backend=_LegacyToolBackend()
+    )
+
+    trial = outcome.report.gate_trials[0]
+    probe = trial["adversarial_probe"]
+    assert probe["n_inconclusive"] == 1 and probe["conclusive"] is False
+    # Advisory evidence never changes the decision; blocking fails closed with
+    # a reason that names the unsupported route rather than a score drop.
+    assert trial["blocked_by_adversarial"] is blocking
+    assert probe["block_reason"] == (
+        "inconclusive_repeated_samples_unsupported" if blocking else ""
+    )
+    with open(
+        os.path.join(outcome.staging_dir, "report.md"), encoding="utf-8"
+    ) as handle:
+        markdown = handle.read()
+    assert (
+        "1 probe(s) inconclusive: their replay route cannot produce distinct "
+        "repeated samples, so they were not scored and cannot flag the candidate."
+    ) in markdown
+    assert "No conclusive probe was scored; blocking mode fails closed" in markdown
+    assert (
+        "| `train` | `train_adversarial_request-frame` | request-frame | "
+        "— | — | — | — | — | inconclusive |"
+    ) in markdown
diff --git a/tests/test_sleep_engine.py b/tests/test_sleep_engine.py
index 3f9753615..289732380 100644
--- a/tests/test_sleep_engine.py
+++ b/tests/test_sleep_engine.py
@@ -2913,6 +2913,10 @@ def test_group_runs_inherit_dream_gate_budget_and_scoped_recall_config(self):
                 recall_k=7,
                 dream_rollouts=3,
                 dream_factor=2,
+                dream_adversarial=2,
+                dream_adversarial_blocking=True,
+                dream_adversarial_margin=0.015,
+                dream_adversarial_rollouts=2,
                 edit_budget=6,
                 gate_metric="mixed",
                 gate_mixed_weight=0.37,
@@ -2953,7 +2957,12 @@ def _spy(backend, tasks, skill, memory, **kwargs):
                 # Keep this regression fast while preserving the exact
                 # arguments observed at the public orchestration boundary.
                 safe = dict(kwargs)
-                safe.update(recall_k=0, dream_rollouts=1, dream_factor=0)
+                safe.update(
+                    recall_k=0,
+                    dream_rollouts=1,
+                    dream_factor=0,
+                    dream_adversarial=0,
+                )
                 return real_dream_consolidate(
                     backend,
                     tasks,
@@ -2988,6 +2997,10 @@ def _spy(backend, tasks, skill, memory, **kwargs):
                 "recall_k": 7,
                 "dream_rollouts": 3,
                 "dream_factor": 2,
+                "dream_adversarial": 2,
+                "dream_adversarial_blocking": True,
+                "dream_adversarial_margin": 0.015,
+                "dream_adversarial_rollouts": 2,
                 "edit_budget": 6,
                 "gate_metric": "mixed",
                 "gate_mixed_weight": 0.37,