Wave 3: close the HIGH review findings (kill-switch wiring, HeldoutSplit, EKS entrypoint bug)

Phase-7 reconciliation of the concurrent review team's findings (research/review-*.json).

HIGH:
- R1: wire HeldOutGuard into ComposerReplicationTrainer — OPTIONAL + OFF by
default (no heldout_guard => byte-identical legacy behavior). When configured,
_maybe_update_killswitch folds in-loop reward + injected heldout_eval_fn() +
token-mean KL into the guard at the logging cadence; fires a hard
(CollapseStopError) or soft (control.should_training_stop) halt. The #2
collapse safeguard now actually fires instead of being dead code.
+ test_killswitch_integration.py.
- R2: build composer_replication/safety/holdout.py — HeldoutSplit disjointness
enforcer (id-set + optional content-hash; .split()/.validate()/.assert_disjoint;
raises HeldoutOverlapError listing the leak). The un-built second half of C1
that keeps the guard's proxy-real gap signal meaningful. +10 tests. Re-exported
from safety/__init__ (resolves the dangling refs).
- R3: EKS entrypoint contract bug — replica_entrypoint.__main__ hard-required
--rendezvous/--world-size/--trainer-module argv, but EKSExecutor passes them
as ENV vars => a real EKS pod crashed at arg-parsing. Fixed: __main__ now
resolves each field from argv OR the matching env var (RENDEZVOUS_URI/
WORLD_SIZE/TRAINER_MODULE), erroring only if NEITHER supplies a required field.
SageMaker's argv path still works. Proven end-to-end with a pure-env invocation.

LOW:
- R4: calibrate_kl_threshold now rejects factor<=0 / negative baseline KL and
floors the result at 1e-6, so calibration can never produce a non-positive
kl_hard_stop (which would fire on every healthy step).
- R10: test pinning path-(c) gap-blowout as an intentional divergence-RATE gate
(fires on fast proxy/real divergence even while real still rises).

DOCS:
- R7: API_REFERENCE §15-17 documenting EKSExecutor, SageMakerExecutor,
DockerSandbox, and the safety module (HeldOutGuard/TripwireStatus/etc.).
- R8: ADR-015 (held-out + kill-switch design) authored + added to the ADR index.

Deferred LOW (tracked in docs/BACKLOG_RESOLUTION): R5 (executor cancel
exception-narrowing), R6 (EKS collect result-key uniformity), R11 (spike-006
flaky-under-contention). safety suite 37/37, trainer 73/73, serverless 53/53.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

README.md

@@ -206,7 +206,7 @@ dimensions. Six new artifact families:
   using the framework to RL-train altered-minds-altered models. ~$300
   estimated for a moral-scenarios trace-replay round.
 
-**Tests as of Wave 15: 115 passing + 1 skip-marked.** Wave-by-wave: 72 (W12) → 93 (W13) → 124 (W14) → 130 (W14b) → 115 (W15: TAID rewrite consolidated 16 schedule-tests into 7 t-paramaterized tests; OPSD parity test added skip-marked). See `docs/V1_V8_COVERAGE.md` for the canonical running count.
+**Tests (canonical, measured 2026-06-09): 266 passing / 62 skipped / 328 collected** — see `docs/V1_V8_COVERAGE.md` for the canonical count and why the skip count varies by environment (optional deps + Docker host). Historical wave-by-wave growth: 72 (W12) → 93 (W13) → 124 (W14) → 130 (W14b) → 115 (W15) → 266 (2026-06).
 
 ## Methodology — how this synthesis was produced
 

composer_replication/diloco/serverless/replica_entrypoint.py

@@ -92,18 +92,50 @@ if __name__ == "__main__":
     import argparse
     import json
 
+    # Dual input contract (both backends supported):
+    #   * argv  — SageMakerExecutor / LocalProcessExecutor pass the run config as
+    #     `--rendezvous/--world-size/--trainer-module` ContainerArguments.
+    #   * env   — EKSExecutor (and any backend that prefers a pure-env contract,
+    #     since k8s Indexed Jobs already inject REPLICA_RANK via the downward API)
+    #     pass the SAME values as RENDEZVOUS_URI / WORLD_SIZE / TRAINER_MODULE
+    #     env vars. The argv flags are therefore NOT `required=True`: when absent
+    #     we fall back to the env vars, and only error if NEITHER source supplies
+    #     a mandatory field. This is the R3 fix — previously the argparse block
+    #     hard-required argv, so an EKS pod (env-only) crashed at arg-parsing.
     parser = argparse.ArgumentParser()
-    parser.add_argument("--rendezvous", required=True)
-    parser.add_argument("--world-size", type=int, required=True)
-    parser.add_argument("--trainer-module", required=True)
-    parser.add_argument("--trainer-fn", default="train")
-    parser.add_argument("--trainer-kwargs-json", default="{}")
+    parser.add_argument("--rendezvous", default=None)
+    parser.add_argument("--world-size", type=int, default=None)
+    parser.add_argument("--trainer-module", default=None)
+    parser.add_argument("--trainer-fn", default=None)
+    parser.add_argument("--trainer-kwargs-json", default=None)
     args = parser.parse_args()
 
+    def _resolve(arg_val, env_key, *, required, cast=lambda x: x):
+        if arg_val is not None:
+            return arg_val
+        env_val = os.environ.get(env_key)
+        if env_val is not None:
+            return cast(env_val)
+        if required:
+            raise SystemExit(
+                f"replica_entrypoint: missing '{env_key}' — supply it via the "
+                f"argv flag or the {env_key} environment variable "
+                f"(EKSExecutor uses env; SageMaker/Local use argv)."
+            )
+        return None
+
+    rendezvous = _resolve(args.rendezvous, "RENDEZVOUS_URI", required=True)
+    world_size = _resolve(args.world_size, "WORLD_SIZE", required=True, cast=int)
+    trainer_module = _resolve(args.trainer_module, "TRAINER_MODULE", required=True)
+    trainer_fn = _resolve(args.trainer_fn, "TRAINER_FN", required=False) or "train"
+    kwargs_json = _resolve(
+        args.trainer_kwargs_json, "TRAINER_KWARGS_JSON", required=False
+    ) or "{}"
+
     main(
-        rendezvous_uri=args.rendezvous,
-        world_size=args.world_size,
-        trainer_module=args.trainer_module,
-        trainer_fn=args.trainer_fn,
-        trainer_kwargs=json.loads(args.trainer_kwargs_json),
+        rendezvous_uri=rendezvous,
+        world_size=world_size,
+        trainer_module=trainer_module,
+        trainer_fn=trainer_fn,
+        trainer_kwargs=json.loads(kwargs_json),
     )

composer_replication/safety/__init__.py

@@ -13,12 +13,19 @@ Public surface:
                      .proxy_real_gap)
   - CollapseStopError   — typed exception for exception-based trainer control flow
   - kl_token_trust_filter — per-token KL trust-region mask (torchrl KL-Mask analog)
+  - HeldoutSplit / HeldoutOverlapError — the train/held-out set-disjointness
+                     enforcer (holdout.py) that keeps the guard's proxy-real gap
+                     signal meaningful (a held-out set that drifts into the train
+                     set makes the gap meaningless).
 
-Pure-Python, no torch / cloud deps. See docs/adrs/ADR-015-*.md and the
-'holdout-killswitch' research digest.
+Pure-Python, no torch / cloud deps. See docs/adrs/ADR-015-holdout-killswitch.md.
 """
 from __future__ import annotations
 
+from composer_replication.safety.holdout import (
+    HeldoutOverlapError,
+    HeldoutSplit,
+)
 from composer_replication.safety.kill_switch import (
     CollapseStopError,
     HeldOutGuard,
@@ -31,4 +38,6 @@ __all__ = [
     "TripwireStatus",
     "CollapseStopError",
     "kl_token_trust_filter",
+    "HeldoutSplit",
+    "HeldoutOverlapError",
 ]

composer_replication/safety/holdout.py

@@ -0,0 +1,354 @@
+"""holdout.py — held-out / train set-disjointness enforcer (the #2 safeguard,
+   second half).
+
+``kill_switch.py`` (the ``HeldOutGuard`` run-level collapse tripwire) is only
+sound if the held-out eval it watches is *genuinely disjoint* from the tasks the
+generator trains on. If a single held-out task leaks back into the train /
+generator pool, the "real" eval drifts WITH the train set and the proxy-real
+Hacking-Gap signal becomes meaningless (see the Shumailov / Gao collapse
+references in ``kill_switch.py``: the held-out eval must stay anchored to REAL
+tasks that are NEVER fed back to the generator). This module enforces that
+discipline mechanically rather than leaving it to convention.
+
+``HeldoutSplit`` enforces disjointness two ways, both pure-Python:
+
+  - **id-based** — the train/generator ``task_id`` set and the held-out
+    ``task_id`` set must not intersect. This is the cheap, exact check.
+
+  - **content-hash-based** (optional, ``check_content=True``) — a sha256 over a
+    *normalized* view of each task's content. This catches NEAR-DUPLICATES that
+    slipped through with DIFFERENT ids: the same broken repo + same
+    ``fail_to_pass`` targets re-minted under a fresh ``task_id`` would pass the
+    id check but is, for collapse purposes, the same eval task leaking into
+    train. The EvilGenie failure-mode literature (arXiv 2511.21654, cited in
+    ``kill_switch.py``) is explicit that "holdout tests have many surprising
+    failure modes" — silent re-id'd duplicates are one of them.
+
+The ``split(all_tasks, holdout_frac, seed)`` constructor produces a
+GUARANTEED-disjoint (train, holdout) partition deterministically: a fixed seed
+yields the same partition every run, so the held-out anchor is reproducible
+across the long self-evolving run.
+
+Pure-Python: only ``hashlib`` / ``random`` from the stdlib. No torch, no cloud
+deps. Accepts either raw ``task_id`` strings OR ``FeatureDeletionTask`` objects
+(anything with a ``task_id`` attribute) on every entry point.
+"""
+from __future__ import annotations
+
+import hashlib
+import random
+from collections.abc import Iterable, Sequence
+from dataclasses import dataclass, field
+from typing import Any
+
+
+class HeldoutOverlapError(ValueError):
+    """Raised when the train/generator pool and the held-out eval pool overlap.
+
+    Carries the offending identifiers so the caller can log exactly which tasks
+    leaked across the boundary (mirroring how ``datagen/monitor.py`` surfaces the
+    specific suspected hacks rather than a bare boolean).
+
+    Attributes:
+        overlapping_ids: sorted task ids present in BOTH pools (id-based leak).
+        overlapping_hashes: sorted content hashes present in both pools with
+            *different* ids (content-based near-duplicate leak); empty unless
+            content-hashing was enabled.
+    """
+
+    def __init__(
+        self,
+        overlapping_ids: Sequence[str] = (),
+        overlapping_hashes: Sequence[str] = (),
+    ) -> None:
+        self.overlapping_ids = tuple(overlapping_ids)
+        self.overlapping_hashes = tuple(overlapping_hashes)
+        parts: list[str] = []
+        if self.overlapping_ids:
+            parts.append(
+                f"{len(self.overlapping_ids)} task id(s) appear in BOTH the "
+                f"train/generator pool and the held-out eval pool: "
+                f"{list(self.overlapping_ids)}"
+            )
+        if self.overlapping_hashes:
+            parts.append(
+                f"{len(self.overlapping_hashes)} content hash(es) collide across "
+                f"the boundary with DIFFERENT ids (re-id'd near-duplicates): "
+                f"{list(self.overlapping_hashes)}"
+            )
+        if not parts:  # defensive — should not be raised with nothing overlapping
+            parts.append("train/held-out overlap detected (no identifiers captured)")
+        super().__init__(
+            "held-out eval is NOT disjoint from the train/generator pool — "
+            "this corrupts the proxy-real collapse signal. " + "; ".join(parts)
+        )
+
+
+def task_id_of(task: Any) -> str:
+    """Coerce a task (a ``task_id`` string or a ``FeatureDeletionTask``-like
+    object with a ``.task_id`` attribute) to its id string.
+
+    Raises:
+        TypeError: if ``task`` is neither a string nor has a ``task_id``.
+    """
+    if isinstance(task, str):
+        return task
+    tid = getattr(task, "task_id", None)
+    if isinstance(tid, str):
+        return tid
+    raise TypeError(
+        f"expected a task_id str or an object with a str .task_id attribute, "
+        f"got {type(task).__name__!r}"
+    )
+
+
+def content_hash(task: Any) -> str:
+    """sha256 over a NORMALIZED view of a task's content (id-independent).
+
+    The hash deliberately EXCLUDES ``task_id`` so two tasks that are identical
+    apart from their id collide — that collision is exactly the near-duplicate
+    leak we want ``check_content=True`` to catch.
+
+    Normalization (so cosmetic differences do not defeat the check):
+      - for ``FeatureDeletionTask``-like objects, hash the load-bearing content
+        fields (repo, base_commit, broken_image, test_command, the SORTED
+        fail_to_pass / pass_to_pass test sets, granularity, sorted
+        deleted_symbols) — NOT task_id, and NOT volatile/advisory fields like
+        difficulty_prior or upstream_license;
+      - for a bare string, hash the whitespace-collapsed, lower-cased text (a
+        plain id string is its own content);
+      - test-set tuples are sorted so reordering the same tests does not change
+        the hash.
+
+    A plain ``task_id`` string therefore hashes to a stable, content-derived
+    value; passing the same strings to both pools will collide on id FIRST
+    (the id check fires before the content check), so the string path is mainly
+    a graceful fallback for callers without structured tasks.
+    """
+    fields = _content_fields(task)
+    blob = "\x1f".join(fields)  # unit-separator join: unambiguous field boundary
+    return hashlib.sha256(blob.encode("utf-8")).hexdigest()
+
+
+def _normalize_text(text: str) -> str:
+    """Collapse runs of whitespace and lower-case, so cosmetic reformatting of a
+    command / repo string does not defeat content-hash matching."""
+    return " ".join(text.split()).lower()
+
+
+def _content_fields(task: Any) -> list[str]:
+    """Ordered, normalized content fields for hashing (id excluded)."""
+    if isinstance(task, str):
+        return [_normalize_text(task)]
+
+    # FeatureDeletionTask-like: pull the content-defining fields if present.
+    def norm(attr: str) -> str:
+        val = getattr(task, attr, None)
+        return _normalize_text(str(val)) if val is not None else ""
+
+    def norm_set(attr: str) -> str:
+        # Sorted so test-order does not change the hash; each test normalized.
+        vals = getattr(task, attr, None) or ()
+        return "\x1e".join(sorted(_normalize_text(str(v)) for v in vals))
+
+    if hasattr(task, "task_id"):
+        return [
+            norm("repo"),
+            norm("base_commit"),
+            norm("broken_image"),
+            norm("test_command"),
+            norm_set("fail_to_pass"),
+            norm_set("pass_to_pass"),
+            norm("granularity"),
+            norm_set("deleted_symbols"),
+        ]
+
+    # Last resort: a non-string, non-task object — hash its repr (best-effort).
+    return [_normalize_text(repr(task))]
+
+
+@dataclass(frozen=True)
+class HeldoutSplit:
+    """A (train/generator, held-out eval) partition with a disjointness contract.
+
+    Construct directly from two iterables of task ids (or
+    ``FeatureDeletionTask`` objects)::
+
+        split = HeldoutSplit(train_tasks, holdout_tasks)
+        split.assert_disjoint()           # raises HeldoutOverlapError on a leak
+        if split.is_disjoint: ...
+
+    or deterministically partition one pool::
+
+        split = HeldoutSplit.split(all_tasks, holdout_frac=0.2, seed=1234)
+
+    Set ``check_content=True`` to also reject re-id'd near-duplicates (same
+    normalized content under a different ``task_id``). Content-hashing is a
+    superset check: a content collision with the SAME id is just the id leak and
+    is reported via ``overlapping_ids``; a collision with DIFFERENT ids is the
+    near-duplicate leak reported via ``overlapping_content_hashes``.
+
+    The instance is frozen; the id/hash sets are computed once at construction.
+    """
+
+    train_ids: frozenset[str]
+    holdout_ids: frozenset[str]
+    check_content: bool = False
+    # content hash -> set of ids, per pool (only populated when check_content).
+    _train_hashes: dict[str, frozenset[str]] = field(default_factory=dict, repr=False)
+    _holdout_hashes: dict[str, frozenset[str]] = field(default_factory=dict, repr=False)
+
+    # ------------------------------------------------------------------------
+    # construction
+    # ------------------------------------------------------------------------
+    def __init__(
+        self,
+        train: Iterable[Any],
+        holdout: Iterable[Any],
+        *,
+        check_content: bool = False,
+    ) -> None:
+        train_list = list(train)
+        holdout_list = list(holdout)
+
+        object.__setattr__(self, "train_ids", frozenset(map(task_id_of, train_list)))
+        object.__setattr__(self, "holdout_ids", frozenset(map(task_id_of, holdout_list)))
+        object.__setattr__(self, "check_content", bool(check_content))
+
+        if check_content:
+            object.__setattr__(self, "_train_hashes", _hash_index(train_list))
+            object.__setattr__(self, "_holdout_hashes", _hash_index(holdout_list))
+        else:
+            object.__setattr__(self, "_train_hashes", {})
+            object.__setattr__(self, "_holdout_hashes", {})
+
+    # ------------------------------------------------------------------------
+    # deterministic constructor
+    # ------------------------------------------------------------------------
+    @classmethod
+    def split(
+        cls,
+        all_tasks: Iterable[Any],
+        holdout_frac: float = 0.2,
+        seed: int = 0,
+        *,
+        check_content: bool = False,
+    ) -> HeldoutSplit:
+        """Deterministically partition ``all_tasks`` into a disjoint (train,
+        held-out) split.
+
+        The partition is keyed on each task's ``task_id`` so it is reproducible
+        across runs (same ``all_tasks`` ids + same ``seed`` => same split). Tasks
+        are de-duplicated by id first (a duplicate id cannot land on both sides),
+        then shuffled with a SEEDED ``random.Random`` and sliced — guaranteeing a
+        disjoint result by construction.
+
+        Args:
+            all_tasks: the full pool (ids or ``FeatureDeletionTask`` objects).
+            holdout_frac: fraction routed to the held-out pool, in [0, 1]. The
+                held-out size is ``round(n * holdout_frac)``, clamped so that a
+                non-empty pool with ``0 < holdout_frac < 1`` always leaves at
+                least one task on EACH side.
+            seed: PRNG seed for the deterministic shuffle.
+            check_content: enable content-hash disjointness on the result too.
+
+        Returns:
+            A ``HeldoutSplit`` whose ``is_disjoint`` is True by construction.
+
+        Raises:
+            ValueError: if ``holdout_frac`` is outside [0, 1].
+        """
+        if not (0.0 <= holdout_frac <= 1.0):
+            raise ValueError(
+                f"holdout_frac must be in [0, 1], got {holdout_frac!r}"
+            )
+
+        # De-dup by id, preserving first-seen order, keeping the original object
+        # so content-hashing (if enabled) sees the structured task.
+        seen: set[str] = set()
+        unique: list[Any] = []
+        for t in all_tasks:
+            tid = task_id_of(t)
+            if tid not in seen:
+                seen.add(tid)
+                unique.append(t)
+
+        n = len(unique)
+        n_holdout = round(n * holdout_frac)
+        # Clamp so a meaningful frac never collapses one side to empty.
+        if 0.0 < holdout_frac < 1.0 and n >= 2:
+            n_holdout = min(max(n_holdout, 1), n - 1)
+
+        # Deterministic shuffle on a COPY (does not mutate caller input).
+        order = list(unique)
+        random.Random(seed).shuffle(order)
+        holdout = order[:n_holdout]
+        train = order[n_holdout:]
+        return cls(train, holdout, check_content=check_content)
+
+    # ------------------------------------------------------------------------
+    # disjointness checks
+    # ------------------------------------------------------------------------
+    def overlapping_ids(self) -> tuple[str, ...]:
+        """Sorted task ids present in BOTH pools (the id-based leak set)."""
+        return tuple(sorted(self.train_ids & self.holdout_ids))
+
+    def overlapping_content_hashes(self) -> tuple[str, ...]:
+        """Sorted content hashes that collide across pools with DIFFERENT ids.
+
+        Empty when ``check_content`` is False. A hash present in both pools whose
+        only shared ids are already plain id-overlaps is not reported here (that
+        leak surfaces via ``overlapping_ids``); only collisions that involve at
+        least one DIFFERENT id on each side count, so the two checks do not
+        double-report the same leak.
+        """
+        if not self.check_content:
+            return ()
+        id_overlap = self.train_ids & self.holdout_ids
+        bad: list[str] = []
+        for h, train_ids in self._train_hashes.items():
+            holdout_ids = self._holdout_hashes.get(h)
+            if holdout_ids is None:
+                continue
+            # Same content on both sides via at least one id that is NOT itself a
+            # plain id-overlap => a re-id'd near-duplicate leak.
+            if (holdout_ids - id_overlap) and (train_ids - id_overlap):
+                bad.append(h)
+        return tuple(sorted(bad))
+
+    @property
+    def is_disjoint(self) -> bool:
+        """True iff the pools share no task id (and, when ``check_content``, no
+        cross-id near-duplicate content)."""
+        if self.train_ids & self.holdout_ids:
+            return False
+        if self.check_content and self.overlapping_content_hashes():
+            return False
+        return True
+
+    def validate(self) -> HeldoutSplit:
+        """Assert disjointness; return ``self`` so it chains in a constructor.
+
+        Raises:
+            HeldoutOverlapError: listing the overlapping ids (and, when
+                ``check_content``, the near-duplicate content hashes).
+        """
+        id_overlap = self.overlapping_ids()
+        hash_overlap = self.overlapping_content_hashes()
+        if id_overlap or hash_overlap:
+            raise HeldoutOverlapError(id_overlap, hash_overlap)
+        return self
+
+    # Documented alias: the task spec names both `validate()` and
+    # `assert_disjoint()` — expose both so either calling convention works.
+    def assert_disjoint(self) -> HeldoutSplit:
+        """Alias for ``validate()`` — raise ``HeldoutOverlapError`` on any leak."""
+        return self.validate()
+
+
+def _hash_index(tasks: Iterable[Any]) -> dict[str, frozenset[str]]:
+    """Map content hash -> frozenset of task ids producing that hash."""
+    acc: dict[str, set[str]] = {}
+    for t in tasks:
+        acc.setdefault(content_hash(t), set()).add(task_id_of(t))
+    return {h: frozenset(ids) for h, ids in acc.items()}

composer_replication/safety/kill_switch.py

@@ -401,20 +401,37 @@ class HeldOutGuard:
 
         Args:
             baseline_kls: per-step token-mean KL values from early in the run.
-            factor: multiplier on the baseline mean (default 3.0).
+                KL is non-negative by definition, so every value must be >= 0.
+            factor: multiplier on the baseline mean. Must be > 0.
 
         Returns:
-            The new ``kl_hard_stop`` (also stored on the instance).
+            The new ``kl_hard_stop`` (also stored on the instance), always > 0.
 
         Raises:
-            ValueError: if ``baseline_kls`` is empty.
+            ValueError: if ``baseline_kls`` is empty, ``factor <= 0``, or any
+                baseline KL is negative.
         """
         if not baseline_kls:
             raise ValueError("baseline_kls must be non-empty to calibrate")
+        # GUARD (R4): a non-positive factor or a negative baseline would make
+        # `calibrated` <= 0, and min(<=0, 0.08) = a NON-POSITIVE kl_hard_stop —
+        # after which the KL tripwire fires on EVERY healthy step (any positive
+        # KL EMA exceeds a non-positive ceiling). KL is non-negative by
+        # definition, so these inputs are nonsensical; reject them loudly rather
+        # than silently disarm-by-inverting the guard.
+        if factor <= 0:
+            raise ValueError(f"factor must be > 0, got {factor!r}")
+        if any(k < 0 for k in baseline_kls):
+            raise ValueError(
+                f"baseline_kls must all be >= 0 (KL is non-negative); got a "
+                f"negative value in {baseline_kls!r}"
+            )
         mean_kl = sum(baseline_kls) / len(baseline_kls)
         calibrated = factor * mean_kl
         # Only tighten: never let calibration loosen past the current ceiling.
-        self.kl_hard_stop = min(calibrated, self.kl_hard_stop)
+        # Floor at a small positive epsilon so an all-zero baseline (mean_kl==0)
+        # can't drive the ceiling to exactly 0 and fire on the first positive KL.
+        self.kl_hard_stop = max(min(calibrated, self.kl_hard_stop), 1e-6)
         return self.kl_hard_stop
 
     # ------------------------------------------------------------------------

composer_replication/safety/tests/test_holdout.py

@@ -0,0 +1,128 @@
+"""Tests for HeldoutSplit — the train/held-out set-disjointness enforcer.
+
+This is the second half of C1 (the first half is HeldOutGuard in kill_switch.py):
+the guard's proxy-real-gap signal is only meaningful if the held-out eval set is
+genuinely DISJOINT from the train/generator set. HeldoutSplit enforces that.
+
+Written during Wave-3 integration (the build agent shipped holdout.py without a
+test module — same test-gap pattern as the SageMaker/EKS executors).
+"""
+from __future__ import annotations
+
+import pytest
+
+from composer_replication.safety import HeldoutOverlapError, HeldoutSplit
+
+
+# A tiny FeatureDeletionTask-like stand-in (HeldoutSplit reads task_id + content
+# fields via duck-typing; a plain object with task_id works, and a string is
+# treated as its own id).
+class _Task:
+    """FeatureDeletionTask-like stand-in. content-hashing in holdout.py reads the
+    real task fields (repo/base_commit/test_command/...), so the content kwarg
+    populates `repo` (one of the hashed fields), not a generic `content` attr."""
+
+    def __init__(self, task_id, content=""):
+        self.task_id = task_id
+        self.repo = content  # `repo` is one of the fields _content_fields hashes
+
+
+# ---------------------------------------------------------------------
+# id-based disjointness
+# ---------------------------------------------------------------------
+
+
+def test_disjoint_passes():
+    s = HeldoutSplit(train=["a", "b", "c"], holdout=["d", "e"])
+    assert s.is_disjoint
+    assert s.overlapping_ids() == ()
+    # validate / assert_disjoint return self and do not raise
+    assert s.validate() is s
+    assert s.assert_disjoint() is s
+
+
+def test_overlap_raises_and_lists_ids():
+    s = HeldoutSplit(train=["a", "b", "shared"], holdout=["shared", "z"])
+    assert not s.is_disjoint
+    assert s.overlapping_ids() == ("shared",)
+    with pytest.raises(HeldoutOverlapError) as exc:
+        s.validate()
+    assert "shared" in str(exc.value)
+
+
+def test_object_tasks_use_task_id():
+    train = [_Task("t1"), _Task("t2")]
+    holdout = [_Task("h1")]
+    assert HeldoutSplit(train, holdout).is_disjoint
+    # an object sharing an id with train is a leak
+    leak = HeldoutSplit(train, [_Task("t1")])
+    assert not leak.is_disjoint
+    assert leak.overlapping_ids() == ("t1",)
+
+
+# ---------------------------------------------------------------------
+# deterministic split()
+# ---------------------------------------------------------------------
+
+
+def test_split_is_disjoint_by_construction():
+    pool = [f"task{i}" for i in range(10)]
+    s = HeldoutSplit.split(pool, holdout_frac=0.3, seed=0)
+    assert s.is_disjoint
+    # every id is on exactly one side; union covers the (de-duped) pool
+    assert s.train_ids.isdisjoint(s.holdout_ids)
+    assert (s.train_ids | s.holdout_ids) == set(pool)
+
+
+def test_split_is_deterministic():
+    pool = [f"task{i}" for i in range(20)]
+    a = HeldoutSplit.split(pool, holdout_frac=0.25, seed=42)
+    b = HeldoutSplit.split(pool, holdout_frac=0.25, seed=42)
+    assert a.holdout_ids == b.holdout_ids
+    assert a.train_ids == b.train_ids
+    # a different seed gives a (very likely) different partition
+    c = HeldoutSplit.split(pool, holdout_frac=0.25, seed=7)
+    assert c.holdout_ids != a.holdout_ids or c.train_ids != a.train_ids
+
+
+def test_split_never_collapses_a_side():
+    s = HeldoutSplit.split([f"t{i}" for i in range(5)], holdout_frac=0.01, seed=0)
+    assert len(s.holdout_ids) >= 1
+    assert len(s.train_ids) >= 1
+
+
+def test_split_rejects_bad_frac():
+    with pytest.raises(ValueError):
+        HeldoutSplit.split(["a", "b"], holdout_frac=1.5)
+
+
+def test_split_dedups_by_id():
+    # duplicate ids cannot land on both sides
+    pool = ["a", "a", "b", "c"]
+    s = HeldoutSplit.split(pool, holdout_frac=0.5, seed=0)
+    assert s.is_disjoint
+    assert (s.train_ids | s.holdout_ids) == {"a", "b", "c"}
+
+
+# ---------------------------------------------------------------------
+# content-hash disjointness (catches same-content / different-id near-dups)
+# ---------------------------------------------------------------------
+
+
+def test_content_hash_catches_same_content_different_id():
+    # different ids, identical content -> id-disjoint but content-leaked
+    train = [_Task("t1", content="fix the off-by-one in range()")]
+    holdout = [_Task("h1", content="fix the off-by-one in range()")]
+    s = HeldoutSplit(train, holdout, check_content=True)
+    assert s.overlapping_ids() == ()  # ids are disjoint
+    assert not s.is_disjoint  # but content collides
+    assert s.overlapping_content_hashes()  # non-empty
+    with pytest.raises(HeldoutOverlapError):
+        s.validate()
+
+
+def test_content_hash_disjoint_when_content_differs():
+    train = [_Task("t1", content="alpha")]
+    holdout = [_Task("h1", content="beta")]
+    s = HeldoutSplit(train, holdout, check_content=True)
+    assert s.is_disjoint

composer_replication/safety/tests/test_kill_switch.py

@@ -318,3 +318,55 @@ def test_kl_token_trust_filter_masks_above_threshold():
     assert kl_token_trust_filter(0.20, threshold=0.08) is True   # too large -> mask
     assert kl_token_trust_filter(0.05, threshold=0.08) is False  # within trust region
     assert kl_token_trust_filter(0.08, threshold=0.08) is False  # boundary, not masked
+
+
+# --- R4: calibrate_kl_threshold input guards (negative factor / baseline) -----
+
+def test_calibrate_rejects_nonpositive_factor():
+    """R4: a factor<=0 would make calibrated<=0 and min(<=0, 0.08)<=0, after
+    which the KL tripwire fires on every healthy step. Reject it loudly."""
+    g = _guard()
+    with pytest.raises(ValueError, match="factor must be > 0"):
+        g.calibrate_kl_threshold([0.01, 0.02], factor=-3.0)
+    with pytest.raises(ValueError, match="factor must be > 0"):
+        g.calibrate_kl_threshold([0.01, 0.02], factor=0.0)
+
+
+def test_calibrate_rejects_negative_baseline_kl():
+    """R4: KL is non-negative by definition; a negative baseline is nonsensical
+    and could invert the ceiling. Reject it."""
+    g = _guard()
+    with pytest.raises(ValueError, match="non-negative"):
+        g.calibrate_kl_threshold([0.01, -0.5, 0.02])
+
+
+def test_calibrate_never_yields_nonpositive_threshold():
+    """R4: even an all-zero baseline (mean 0) must leave a positive ceiling so a
+    later positive KL doesn't fire spuriously."""
+    g = _guard()
+    out = g.calibrate_kl_threshold([0.0, 0.0, 0.0], factor=3.0)
+    assert out > 0.0
+    assert g.kl_hard_stop > 0.0
+
+
+# --- R10: path-(c) gap-blowout is a divergence-RATE gate, not a real-decline --
+
+def test_gap_blowout_fires_even_when_real_still_rising():
+    """R10: path (c) fires when the proxy gain outpaces the real gain beyond the
+    ceiling EVEN WHILE the held-out (real) score is still genuinely RISING. This
+    is INTENTIONAL — path (c) is a divergence-RATE gate (fast single-generation
+    hacking), distinct from path (a)'s real-decline streak. Locking the intended
+    behavior so a future change can't silently turn it into a real-decline gate."""
+    g = _guard(max_proxy_real_gap=0.1, decline_patience=99)  # isolate path (c) from (a)
+    status = None
+    for i in range(8):
+        status = g.update(
+            i,
+            in_loop_reward=0.30 + 0.20 * i,   # proxy sprints
+            heldout_score=0.30 + 0.01 * i,    # real still rising, but slowly
+            kl_to_init=0.02,
+        )
+    assert status.fire, "path (c) should fire on a fast proxy/real divergence"
+    assert "gap" in status.reason.lower()
+    # And the real score WAS rising the whole time (not a decline-driven fire).
+    assert status.heldout_ema > g._fold(None, 0.30)  # type: ignore[attr-defined]

composer_replication/trainer/composer_trainer.py

@@ -26,10 +26,14 @@ The data collator (data_collator.py) is responsible for:
 from __future__ import annotations
 
 import logging
-from typing import Any
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any
 
 import torch
-import torch.nn.functional as F
+import torch.nn.functional as F  # noqa: N812 — repo-wide torch convention
+
+if TYPE_CHECKING:  # type-only — never imported at runtime (keeps the dep lazy)
+    from composer_replication.safety import HeldOutGuard
 
 # These imports work when TRL is installed — they're not skeleton imports.
 # When TRL is missing we fall back to `object` so the module still imports
@@ -63,6 +67,25 @@ class ComposerReplicationTrainer(GRPOTrainer):  # type: ignore[misc, valid-type]
         sdpo_temperature: temperature for SDPO loss; SDPO paper uses 1.0.
         sdpo_token_clip: per-token JSD clip for stability; None = no clip.
         replay_dpo_beta: beta param of the DPO loss (β in the standard DPO formula).
+        heldout_guard: optional ``HeldOutGuard`` (the #2 collapse safeguard from
+            ``composer_replication.safety``). Default None = OFF (no behavior
+            change whatsoever). When supplied, the trainer folds one checkpoint's
+            metrics into the guard at the ``args.logging_steps`` cadence (the same
+            place the loss components are logged) and HALTS the run on a fired
+            verdict — the run-level reward-hacking / collapse tripwire actually
+            firing instead of sitting inert.
+        heldout_eval_fn: zero-arg callable returning the held-out (real) eval
+            score as a float, evaluated each guard cadence. Injectable so the
+            trainer never hardcodes an eval — pass a closure over your disjoint
+            held-out pool (the ``HeldoutSplit`` discipline). Required whenever
+            ``heldout_guard`` is set; the guard's whole signal is in-loop reward
+            vs. this held-out score.
+        strict_killswitch: when True (default), a fired guard verdict raises
+            ``CollapseStopError`` to hard-stop training (exception-based control
+            flow, matching ``HeldOutGuard.raise_if_fired``). When False the
+            verdict is logged and ``self.control.should_training_stop`` is set so
+            the HF loop ends gracefully after the step (soft stop). Only consulted
+            when ``heldout_guard`` is set.
     """
 
     def __init__(
@@ -75,6 +98,9 @@ class ComposerReplicationTrainer(GRPOTrainer):  # type: ignore[misc, valid-type]
         sdpo_token_clip: float | None = None,
         replay_dpo_beta: float = 0.1,
         strict_sdpo_alignment: bool = True,
+        heldout_guard: HeldOutGuard | None = None,
+        heldout_eval_fn: Callable[[], float] | None = None,
+        strict_killswitch: bool = True,
         **kwargs: Any,
     ):
         if not _TRL_AVAILABLE:
@@ -95,6 +121,21 @@ class ComposerReplicationTrainer(GRPOTrainer):  # type: ignore[misc, valid-type]
         # trust-gap flagged in ADR-008). Set False only for production runs
         # where a single malformed batch should warn-and-skip rather than abort.
         self.strict_sdpo_alignment = strict_sdpo_alignment
+        # --- run-level collapse kill-switch (#2 safeguard) -------------------
+        # OPTIONAL + OFF BY DEFAULT: when heldout_guard is None the loss path is
+        # byte-for-byte the legacy behavior. When set, _maybe_update_killswitch
+        # folds metrics into the guard at the logging cadence (see _compute_loss).
+        self.heldout_guard = heldout_guard
+        self.heldout_eval_fn = heldout_eval_fn
+        self.strict_killswitch = strict_killswitch
+        if heldout_guard is not None and heldout_eval_fn is None:
+            raise ValueError(
+                "heldout_guard was provided without heldout_eval_fn: the guard's "
+                "tripwire compares in-loop reward against a DISJOINT held-out "
+                "(real) eval score, so it needs an injectable zero-arg "
+                "heldout_eval_fn() -> float. Pass a closure over your held-out "
+                "pool (the HeldoutSplit discipline)."
+            )
 
     # ----------------------------------------------------------------------
     # Loss override (the integration core)
@@ -130,9 +171,107 @@ class ComposerReplicationTrainer(GRPOTrainer):  # type: ignore[misc, valid-type]
                     "loss/alpha_sdpo":         self.alpha_sdpo,
                     "loss/beta_replay":        self.beta_replay,
                 })
+                # Fold one checkpoint into the run-level collapse kill-switch at
+                # the SAME cadence (no-op unless a guard was configured).
+                self._maybe_update_killswitch()
 
         return total
 
+    # ----------------------------------------------------------------------
+    # Run-level collapse kill-switch (#2 safeguard) — optional, OFF by default
+    # ----------------------------------------------------------------------
+
+    def _maybe_update_killswitch(self) -> None:
+        """Fold this checkpoint's metrics into ``heldout_guard`` and act on a fire.
+
+        No-op when no guard was configured (the default) — this is the
+        backward-compat guarantee: without ``heldout_guard`` the trainer behaves
+        exactly as before. When a guard IS set:
+
+          * ``in_loop_reward`` is the GRPO reward signal TRL already aggregates
+            into ``self._metrics[mode]["reward"]`` each step (we read the latest;
+            no extra forward pass).
+          * ``heldout_score`` comes from the injected ``heldout_eval_fn()`` — the
+            trainer never hardcodes an eval.
+          * ``kl_to_init`` (token-mean nats/token, the ``token_mean_kl``
+            convention the guard expects) is read from TRL's logged ``"kl"``
+            metric when present, else left None (KL path stays inert).
+
+        On a fired verdict the verdict is logged. If ``strict_killswitch`` (the
+        default) the verdict is converted into a ``CollapseStopError`` via
+        ``HeldOutGuard.raise_if_fired`` (hard stop); otherwise the HF training
+        loop is asked to stop gracefully after this step.
+        """
+        guard = self.heldout_guard
+        if guard is None:
+            return  # OFF by default — zero behavior change
+
+        round_idx = int(getattr(self.state, "global_step", 0))
+        in_loop_reward = self._latest_metric("reward")
+        if in_loop_reward is None:
+            # No reward aggregated yet (e.g. very first micro-step before TRL has
+            # populated its metrics). Skip this cadence rather than feed a
+            # fabricated 0.0 that would pollute the guard's baseline/EMA.
+            logger.debug(
+                "kill-switch: no in-loop reward metric yet at step %d; skipping.",
+                round_idx,
+            )
+            return
+
+        assert self.heldout_eval_fn is not None  # enforced in __init__
+        heldout_score = float(self.heldout_eval_fn())
+        kl_to_init = self._latest_metric("kl")  # token-mean KL, or None
+
+        status = guard.update(
+            round_idx=round_idx,
+            in_loop_reward=in_loop_reward,
+            heldout_score=heldout_score,
+            kl_to_init=kl_to_init,
+        )
+
+        self.log({  # type: ignore[attr-defined]
+            "killswitch/in_loop_reward":  status.in_loop_ema,
+            "killswitch/heldout_score":   status.heldout_ema,
+            "killswitch/proxy_real_gap":  status.proxy_real_gap,
+            "killswitch/fire":            float(status.fire),
+        })
+
+        if status.fire:
+            logger.error(
+                "HeldOutGuard FIRED at step %d — halting run. reason: %s",
+                round_idx, status.reason,
+            )
+            if self.strict_killswitch:
+                # Typed exception — exception-based hard stop.
+                guard.raise_if_fired(status)
+            else:
+                # Soft stop: let the HF loop terminate gracefully after this step.
+                control = getattr(self, "control", None)
+                if control is not None:
+                    control.should_training_stop = True
+
+    def _latest_metric(self, name: str) -> float | None:
+        """Most-recent value of a TRL-aggregated train metric, or None.
+
+        TRL's GRPOTrainer appends per-step aggregates to
+        ``self._metrics["train"][name]`` (e.g. ``"reward"``, ``"kl"``). We read
+        the tail defensively so a TRL internals rename degrades to None (KL/reward
+        path goes inert) rather than crashing training.
+        """
+        metrics = getattr(self, "_metrics", None)
+        if not isinstance(metrics, dict):
+            return None
+        train = metrics.get("train")
+        if not isinstance(train, dict):
+            return None
+        series = train.get(name)
+        if not series:
+            return None
+        try:
+            return float(series[-1])
+        except (TypeError, ValueError, IndexError):
+            return None
+
     # ----------------------------------------------------------------------
     # Channel 2: SDPO hint-distill
     # ----------------------------------------------------------------------

composer_replication/trainer/tests/test_killswitch_integration.py

@@ -0,0 +1,370 @@
+"""R1 — HeldOutGuard wired into ComposerReplicationTrainer (the #2 safeguard).
+
+These tests close the "tripwire exists but never fires" gap: the run-level
+collapse kill-switch (``composer_replication.safety.HeldOutGuard``) must actually
+be folded by the trainer at its logging cadence, fed the in-loop GRPO reward and
+an INJECTED held-out eval, and halt the run on a fired verdict.
+
+Acceptance gates:
+  1. BACKWARD-COMPAT: no ``heldout_guard`` => ``_maybe_update_killswitch`` is a
+     pure no-op (never touches the eval fn, never logs) — identical behavior.
+  2. THE WIRING: a fake ``heldout_eval_fn`` that DECLINES while the in-loop
+     reward RISES drives the guard to fire; strict mode raises
+     ``CollapseStopError`` and the verdict carries the reward-hacking signature.
+  3. Soft stop: ``strict_killswitch=False`` => no raise, the HF loop is asked to
+     stop (``control.should_training_stop``).
+  4. Healthy run (held-out tracks reward) never fires.
+  5. KL-to-init is read from TRL's logged metric and reaches the guard.
+  6. Constructor contract: ``heldout_guard`` without ``heldout_eval_fn`` raises.
+  7. ``_compute_loss`` only folds the guard at the ``logging_steps`` cadence
+     (not every micro-step), mirroring the loss-component logging.
+
+CPU-only, no model download, no full GRPOTrainer init (stub instance via
+``__new__`` + manual attribute wiring, the pattern used by the SDPO tests).
+"""
+from __future__ import annotations
+
+import pytest
+
+from composer_replication.safety import CollapseStopError, HeldOutGuard
+from composer_replication.trainer.composer_trainer import ComposerReplicationTrainer
+
+# ---------------------------------------------------------------------------
+# Stubs — mirror _make_sdpo_trainer in test_sdpo_alignment_indices.py: build the
+# trainer via __new__ so we never run GRPOTrainer.__init__ (no TRL setup, no
+# model download), then wire only the attributes the kill-switch path reads.
+# ---------------------------------------------------------------------------
+
+class _State:
+    def __init__(self, global_step: int = 0) -> None:
+        self.global_step = global_step
+
+
+class _Args:
+    def __init__(self, logging_steps: int = 1) -> None:
+        self.logging_steps = logging_steps
+
+
+class _Control:
+    def __init__(self) -> None:
+        self.should_training_stop = False
+
+
+def _make_killswitch_trainer(
+    guard: HeldOutGuard | None,
+    eval_fn,
+    *,
+    strict: bool = True,
+    reward: float | None = 0.40,
+    kl: float | None = None,
+):
+    """A ComposerReplicationTrainer stub exposing only what the kill-switch reads.
+
+    ``reward`` / ``kl`` seed TRL's per-step metric series (the trainer reads the
+    tail of ``self._metrics["train"][name]``). Pass reward=None to simulate "no
+    reward aggregated yet".
+    """
+    obj = ComposerReplicationTrainer.__new__(ComposerReplicationTrainer)
+    obj.heldout_guard = guard
+    obj.heldout_eval_fn = eval_fn
+    obj.strict_killswitch = strict
+    obj.state = _State(global_step=0)
+    obj.args = _Args(logging_steps=1)
+    obj.control = _Control()
+    train_metrics: dict[str, list] = {}
+    if reward is not None:
+        train_metrics["reward"] = [reward]
+    if kl is not None:
+        train_metrics["kl"] = [kl]
+    obj._metrics = {"train": train_metrics}
+    obj.logged: list[dict] = []
+    # capture self.log(...) instead of routing through HF Trainer.log
+    obj.log = obj.logged.append  # type: ignore[assignment]
+    return obj
+
+
+def _set_step_reward(obj, step: int, reward: float, kl: float | None = None) -> None:
+    obj.state.global_step = step
+    obj._metrics["train"].setdefault("reward", []).append(reward)
+    if kl is not None:
+        obj._metrics["train"].setdefault("kl", []).append(kl)
+
+
+# ---------------------------------------------------------------------------
+# Gate 1 — backward compatibility: absent guard is a pure no-op
+# ---------------------------------------------------------------------------
+
+def test_absent_guard_is_noop():
+    """No heldout_guard => the kill-switch path does nothing: the eval fn is
+    never called, nothing is logged, no exception. This is the backward-compat
+    guarantee (no kwarg => identical behavior)."""
+    calls = {"n": 0}
+
+    def eval_fn() -> float:
+        calls["n"] += 1
+        return 0.0
+
+    # Pass eval_fn but NO guard — the helper must never reach the eval fn.
+    obj = _make_killswitch_trainer(guard=None, eval_fn=eval_fn)
+    for step in range(50):
+        _set_step_reward(obj, step, reward=0.40 + 0.05 * step)
+        obj._maybe_update_killswitch()  # must be a no-op
+
+    assert calls["n"] == 0, "held-out eval fn was called even though no guard set"
+    assert obj.logged == [], "kill-switch logged even though no guard configured"
+    assert obj.control.should_training_stop is False
+
+
+def test_constructor_defaults_leave_killswitch_off():
+    """The constructor defaults: a trainer built without the kwargs has
+    heldout_guard=None / strict_killswitch defaulting on but inert."""
+    obj = ComposerReplicationTrainer.__new__(ComposerReplicationTrainer)
+    # Simulate the default-kwarg assignment __init__ performs.
+    obj.heldout_guard = None
+    obj.heldout_eval_fn = None
+    obj.strict_killswitch = True
+    obj._maybe_update_killswitch()  # no state needed: returns immediately on None
+
+
+# ---------------------------------------------------------------------------
+# Gate 2 — THE WIRING: declining held-out + rising reward => guard fires & raises
+# ---------------------------------------------------------------------------
+
+def test_guard_fires_and_raises_on_reward_hacking_signature():
+    """Fake heldout_eval_fn DECLINES while the in-loop reward RISES — the
+    canonical reward-hacking signature. The wired guard must fire and (strict
+    mode) raise CollapseStopError with the reward-hacking reason."""
+    # min_steps small so the test is fast; isolate the decline-streak path.
+    guard = HeldOutGuard(
+        min_steps=3, decline_patience=3, ema_alpha=0.5, max_proxy_real_gap=10.0
+    )
+
+    # held-out declines every call; reward (TRL metric) rises every step.
+    heldout = {"v": 0.80}
+
+    def declining_eval() -> float:
+        heldout["v"] -= 0.05
+        return heldout["v"]
+
+    obj = _make_killswitch_trainer(guard, declining_eval, strict=True)
+
+    raised = None
+    for step in range(1, 30):
+        # rising in-loop reward fed via the TRL metric tail
+        _set_step_reward(obj, step, reward=0.30 + 0.03 * step)
+        try:
+            obj._maybe_update_killswitch()
+        except CollapseStopError as exc:
+            raised = exc
+            break
+
+    assert raised is not None, "guard never fired on the reward-hacking signature"
+    assert guard.should_halt()
+    assert raised.status.fire
+    # The fired verdict must be the held-out-declines-while-reward-rises signature.
+    assert "held-out" in raised.status.reason
+    assert raised.status.proxy_real_gap > 0.0  # proxy gained while real lost
+    # And the kill-switch logged the verdict before raising.
+    assert any("killswitch/fire" in d for d in obj.logged)
+
+
+def test_soft_stop_sets_control_instead_of_raising():
+    """strict_killswitch=False => a fired verdict does NOT raise; it sets the HF
+    loop's control.should_training_stop so training ends gracefully."""
+    guard = HeldOutGuard(
+        min_steps=3, decline_patience=3, ema_alpha=0.5, max_proxy_real_gap=10.0
+    )
+    heldout = {"v": 0.80}
+
+    def declining_eval() -> float:
+        heldout["v"] -= 0.05
+        return heldout["v"]
+
+    obj = _make_killswitch_trainer(guard, declining_eval, strict=False)
+
+    for step in range(1, 30):
+        _set_step_reward(obj, step, reward=0.30 + 0.03 * step)
+        obj._maybe_update_killswitch()  # must NOT raise
+        if obj.control.should_training_stop:
+            break
+
+    assert obj.control.should_training_stop is True, (
+        "soft-stop guard fired but did not request training stop"
+    )
+    assert guard.should_halt()
+
+
+# ---------------------------------------------------------------------------
+# Gate 4 — healthy run never fires
+# ---------------------------------------------------------------------------
+
+def test_healthy_run_never_fires():
+    """Held-out tracks the in-loop reward (both rise together), KL in band =>
+    the wired guard never fires and training is never asked to stop."""
+    guard = HeldOutGuard(
+        min_steps=3, decline_patience=3, ema_alpha=0.5, kl_hard_stop=0.08,
+        max_proxy_real_gap=10.0,
+    )
+    heldout = {"v": 0.28}
+
+    def rising_eval() -> float:
+        heldout["v"] += 0.01
+        return heldout["v"]
+
+    obj = _make_killswitch_trainer(guard, rising_eval, strict=True, kl=0.03)
+
+    for step in range(1, 40):
+        _set_step_reward(obj, step, reward=0.30 + 0.01 * step, kl=0.03)
+        obj._maybe_update_killswitch()  # must never raise on a healthy run
+
+    assert not guard.should_halt()
+    assert obj.control.should_training_stop is False
+
+
+# ---------------------------------------------------------------------------
+# Gate 5 — KL-to-init from TRL's logged metric reaches the guard
+# ---------------------------------------------------------------------------
+
+def test_kl_to_init_is_forwarded_to_guard():
+    """The KL the trainer reads from TRL's "kl" metric must reach the guard's
+    kl_ema (proves kl_to_init wiring), and a KL breach fires via the KL path."""
+    guard = HeldOutGuard(
+        min_steps=3, decline_patience=100, ema_alpha=0.5, kl_hard_stop=0.08,
+        max_proxy_real_gap=10.0,  # isolate the KL path
+    )
+
+    obj = _make_killswitch_trainer(guard, lambda: 0.40, strict=True, kl=0.04)
+
+    # Warm-up with healthy KL; metrics flat so only the KL path can fire.
+    for step in range(1, 5):
+        _set_step_reward(obj, step, reward=0.40, kl=0.04)
+        obj._maybe_update_killswitch()
+    assert guard.last_status is not None and guard.last_status.kl_ema is not None, (
+        "kl_to_init never reached the guard — KL wiring is broken"
+    )
+
+    # KL spikes above the hard stop; EMA climbs and crosses => fire via KL path.
+    raised = None
+    for step in range(5, 20):
+        _set_step_reward(obj, step, reward=0.40, kl=0.30)
+        try:
+            obj._maybe_update_killswitch()
+        except CollapseStopError as exc:
+            raised = exc
+            break
+    assert raised is not None, "KL hard-stop never fired through the wired guard"
+    assert "kl_to_init" in raised.status.reason
+
+
+def test_no_reward_metric_yet_skips_cleanly():
+    """Before TRL has aggregated any reward (empty metric series), the helper
+    skips the fold rather than feeding a fabricated 0.0 into the guard's EMA."""
+    guard = HeldOutGuard(min_steps=3, ema_alpha=0.5)
+    calls = {"n": 0}
+
+    def eval_fn() -> float:
+        calls["n"] += 1
+        return 0.40
+
+    obj = _make_killswitch_trainer(guard, eval_fn, reward=None)
+    obj._maybe_update_killswitch()  # no reward series => skip
+
+    assert calls["n"] == 0, "eval fn called despite no in-loop reward yet"
+    assert guard.last_status is None, "guard advanced despite no reward metric"
+
+
+# ---------------------------------------------------------------------------
+# Gate 6 — constructor contract
+# ---------------------------------------------------------------------------
+
+def test_guard_without_eval_fn_raises_at_construction(monkeypatch):
+    """A guard with no held-out eval is meaningless (the tripwire needs the
+    held-out signal) => the REAL __init__ must reject it loudly. We stub the
+    GRPOTrainer parent __init__ so the validation clause runs without a full
+    TRL/model setup."""
+    parent = ComposerReplicationTrainer.__bases__[0]
+    monkeypatch.setattr(parent, "__init__", lambda self, *a, **k: None, raising=False)
+
+    guard = HeldOutGuard(min_steps=3)
+    with pytest.raises(ValueError, match="heldout_eval_fn"):
+        ComposerReplicationTrainer(heldout_guard=guard)  # no heldout_eval_fn
+
+
+def test_guard_with_eval_fn_constructs_and_stays_off_when_absent(monkeypatch):
+    """The real __init__ wires the kill-switch attributes; with both provided it
+    constructs cleanly, and with neither provided the guard stays None (the
+    default = OFF backward-compat path)."""
+    parent = ComposerReplicationTrainer.__bases__[0]
+    monkeypatch.setattr(parent, "__init__", lambda self, *a, **k: None, raising=False)
+
+    # Both provided => constructs, guard wired.
+    guard = HeldOutGuard(min_steps=3)
+    t = ComposerReplicationTrainer(heldout_guard=guard, heldout_eval_fn=lambda: 0.4)
+    assert t.heldout_guard is guard
+    assert t.strict_killswitch is True  # strict default
+
+    # Neither provided => guard stays None (OFF).
+    t2 = ComposerReplicationTrainer()
+    assert t2.heldout_guard is None
+    assert t2.heldout_eval_fn is None
+
+
+# ---------------------------------------------------------------------------
+# Gate 7 — _compute_loss only folds the guard at the logging cadence
+# ---------------------------------------------------------------------------
+
+def test_compute_loss_folds_guard_only_at_logging_cadence(monkeypatch):
+    """Drive _compute_loss end-to-end (with the GRPO parent loss + SDPO/replay
+    channels stubbed) and assert the guard is folded ONLY on logging-cadence
+    steps — i.e. the kill-switch fold sits inside the same cadence gate as the
+    loss-component logging, not on every micro-step."""
+    import torch
+
+    folds = {"n": 0}
+    guard = HeldOutGuard(min_steps=10_000, ema_alpha=0.5)  # never fires in this test
+
+    def counting_eval() -> float:
+        folds["n"] += 1
+        return 0.40
+
+    obj = ComposerReplicationTrainer.__new__(ComposerReplicationTrainer)
+    obj.alpha_sdpo = 0.0
+    obj.beta_replay = 0.0
+    obj.heldout_guard = guard
+    obj.heldout_eval_fn = counting_eval
+    obj.strict_killswitch = True
+    obj.state = _State(global_step=0)
+    obj.args = _Args(logging_steps=10)
+    obj.control = _Control()
+    obj._metrics = {"train": {"reward": [0.40]}}
+    obj.logged = []
+    obj.log = obj.logged.append  # type: ignore[assignment]
+
+    # Stub the GRPO parent loss (the real `super()._compute_loss` would need a
+    # full TRL trainer) and the SDPO / replay channels to zero. We patch the
+    # PARENT class's _compute_loss so `super()._compute_loss(...)` resolves to it.
+    parent = ComposerReplicationTrainer.__bases__[0]
+    monkeypatch.setattr(
+        parent, "_compute_loss",
+        lambda self, model, inputs: torch.tensor(1.0),
+        raising=False,
+    )
+    monkeypatch.setattr(
+        ComposerReplicationTrainer, "_compute_sdpo_loss",
+        lambda self, model, inputs: torch.tensor(0.0),
+        raising=True,
+    )
+    monkeypatch.setattr(
+        ComposerReplicationTrainer, "_compute_trace_replay_loss",
+        lambda self, model, inputs: torch.tensor(0.0),
+        raising=True,
+    )
+
+    for step in range(0, 35):
+        obj.state.global_step = step
+        obj._metrics["train"]["reward"].append(0.40 + 0.001 * step)
+        total = obj._compute_loss(model=object(), inputs={})
+        assert float(total.detach()) == pytest.approx(1.0)
+
+    # steps 0, 10, 20, 30 are the only cadence hits => exactly 4 guard folds.
+    assert folds["n"] == 4, f"expected 4 cadence folds, got {folds['n']}"

docs/API_REFERENCE.md

@@ -23,6 +23,9 @@ Complete reference for every public symbol in `composer_replication`. Source-of-
 12. `composer_replication.diloco.serverless` (+ `.executor`, `.allreduce`, `.modal`, `.hf_jobs`, `.replica_entrypoint`)
 13. `composer_replication.recipes.prime_rl.composer_loss`
 14. `composer_replication.recipes.monarch.actors`
+15. `composer_replication.diloco.serverless` — cloud executors (`.eks`, `.sagemaker`)
+16. `composer_replication.datagen.docker_sandbox`
+17. `composer_replication.safety` (+ `.kill_switch`)
 
 ---
 
@@ -1498,6 +1501,358 @@ from composer_replication.recipes.monarch.actors import TrainerActor
 
 ---
 
+## 15. `composer_replication.diloco.serverless` — cloud executors
+
+ADR-005 production cloud executors that implement the `ServerlessExecutor` Protocol (see §12) against real clouds. Both are the loud-success siblings of the `ModalExecutor` / `HFJobsExecutor` 🟡 skeletons: cross-replica communication is EXCLUSIVELY the S3 `ObjectStoreAllReduce` rendezvous, so the trainer / loss / DiLoCo math stay byte-for-byte identical regardless of backend. Both set `supports_inter_replica_network = False` (replicas rendezvous only through the object store) and both lazy-import their cloud SDK so `import composer_replication.diloco.serverless` is free of `kubernetes` / `boto3`.
+
+**Extras**: `EKSExecutor` needs `kubernetes>=29` (`pip install -e .[eks]`); `SageMakerExecutor` needs `boto3>=1.34` (`pip install -e .[aws]`). The `[serverless]` extra pulls both (plus `fsspec` + `huggingface_hub` for the rendezvous backends). The SDK is required only at adapter-init / method time — when an api/client is injected (tests), a missing top-level package is tolerated.
+
+### `class EKSExecutor` — `serverless.eks`
+
+```python
+class EKSExecutor:
+    backend_name = "eks"
+    supports_inter_replica_network = False  # S3 rendezvous only
+
+    def __init__(
+        self,
+        image: str,
+        *,
+        namespace: str = "default",
+        service_account_name: str | None = None,
+        node_selector: dict[str, str] | None = None,
+        tolerations: list[Any] | None = None,
+        runtime_class_name: str | None = None,
+        command: list[str] | None = None,
+        cpu_request: str = "4",
+        memory_request: str = "16Gi",
+        ttl_seconds_after_finished: int = 3600,
+        backoff_limit: int = 0,
+        gpu_resource_key: str = "nvidia.com/gpu",
+        run_id: str | None = None,
+        batch_api: Any = None,
+        core_api: Any = None,
+    ) -> None: ...
+    # implements ServerlessExecutor protocol (launch_replicas/poll/stream_logs/cancel/collect)
+```
+
+Run N DiLoCo replicas as a **single Kubernetes Indexed Job** on EKS (`completions == parallelism == n_replicas`, `completionMode="Indexed"`). The control plane assigns each pod a `JOB_COMPLETION_INDEX` 0..N-1 which IS the rank; the executor bridges it to the repo entrypoint's `REPLICA_RANK` env var via the downward API, so `replica_entrypoint` works unchanged. `launch_replicas` creates ONE Job but still returns N `ReplicaHandle`s (`handles[i].rank == i`) sharing the same `job_name`/`namespace` (gang semantics).
+
+**Constructor parameters**
+
+| Name | Type | Default | Meaning |
+|---|---|---|---|
+| `image` | `str` | — | Container image with `composer_replication` installed; runs the replica entrypoint. |
+| `namespace` | `str` (kw-only) | `"default"` | k8s namespace for the Job. |
+| `service_account_name` | `str \| None` (kw-only) | `None` | ServiceAccount referenced on the PodSpec for IRSA / EKS Pod Identity S3 access. REFERENCED only — never created. |
+| `node_selector` | `dict[str,str] \| None` (kw-only) | `None` | Extra node selector merged under the GPU node selector. |
+| `tolerations` | `list[Any] \| None` (kw-only) | `None` | PodSpec tolerations; a `nvidia.com/gpu` NoSchedule toleration is auto-added on GPU requests when none supplied. |
+| `runtime_class_name` | `str \| None` (kw-only) | `None` | Pre-existing RuntimeClass (`"gvisor"` / `"kata"`). Combining with `gpu` is advanced/unverified (see source warning). |
+| `command` | `list[str] \| None` (kw-only) | `None` ⇒ `["python","-m","composer_replication.diloco.serverless.replica_entrypoint"]` | Container command. |
+| `cpu_request` / `memory_request` | `str` (kw-only) | `"4"` / `"16Gi"` | PodSpec resource requests. |
+| `ttl_seconds_after_finished` | `int` (kw-only) | `3600` | Auto-delete the finished Job (cascading) after this many seconds. |
+| `backoff_limit` | `int` (kw-only) | `0` | Job retry budget (fail-fast; NOT the k8s default of 6). |
+| `gpu_resource_key` | `str` (kw-only) | `"nvidia.com/gpu"` | GPU resource key. |
+| `run_id` | `str \| None` (kw-only) | `None` ⇒ `"diloco"` | Prefix baked into the generated Job name. |
+| `batch_api` / `core_api` | `Any` (kw-only) | `None` | DI'd `BatchV1Api` / `CoreV1Api`; lazily built (in-cluster then kube-config) when `None`. Tests inject mocks. |
+
+**`ServerlessExecutor` Protocol methods** (see §12 for the full Protocol)
+
+- `launch_replicas(n_replicas, entrypoint, entrypoint_args, *, gpu=None, timeout=3600) -> list[ReplicaHandle]` — creates ONE Indexed Job; `entrypoint` is ignored (k8s runs the fixed container command), scalar `entrypoint_args` (e.g. `rendezvous_uri`) are passed as upper-cased literal env vars, `gpu` maps to a `nvidia.com/gpu` limit + node selector via the `_GPU_SPEC_TABLE` (`"A100"`/`"H100"`/`"A10G"`/`"T4"`), and `timeout` becomes the Job's `active_deadline_seconds` hard wall-clock kill.
+- `poll(handle) -> str` — reads the single Job status and maps THIS rank: `rank in completed_indexes` ⇒ `"succeeded"`, `rank in failed_indexes` ⇒ `"failed"`, whole-Job `Failed` condition ⇒ `"failed"`, `active > 0` ⇒ `"running"`, else `"pending"`; a 404 (Job gone) ⇒ `"cancelled"`.
+- `stream_logs(handle, *, n_lines=200) -> str` — finds the pod by completion-index annotation/label (or the `<job>-<rank>-` name prefix), tails its log; returns a placeholder string (never raises) when the pod has not started.
+- `cancel(handle) -> None` — deletes the WHOLE shared Indexed Job (`propagation_policy="Background"` so pods are cascadingly deleted, not orphaned). Intentional gang teardown; idempotent (404 swallowed, unknown handle is a no-op).
+- `collect(handles, *, timeout=None) -> list[dict]` — polls (sleeping between, status is eventually consistent) until every rank is terminal or the deadline; per-rank dict is `{"rank","status","exit_code","error","job_name"}` (`exit_code` 0/1/`None`).
+
+**Raises** `RuntimeError` at construction if the `kubernetes` client is absent AND no api was injected; `ValueError` if `n_replicas < 1`.
+
+```python
+from composer_replication.diloco.serverless.eks import EKSExecutor
+ex = EKSExecutor(image="ghcr.io/me/diloco:latest",
+                 service_account_name="diloco-s3")
+handles = ex.launch_replicas(
+    n_replicas=4,
+    entrypoint="",  # ignored on k8s
+    entrypoint_args={"rendezvous_uri": "s3://bucket/run/",
+                     "trainer_module": "my.trainer", "world_size": 4},
+    gpu="H100",
+)
+results = ex.collect(handles, timeout=3600)
+```
+
+### `class SageMakerExecutor` — `serverless.sagemaker`
+
+```python
+class SageMakerExecutor:
+    backend_name = "sagemaker"
+    supports_inter_replica_network = False  # S3 rendezvous only
+
+    def __init__(
+        self,
+        *,
+        role_arn: str,
+        image_uri: str,
+        output_s3_path: str,
+        instance_type: str = "ml.g5.2xlarge",
+        cpu_instance_type: str = "ml.m5.xlarge",
+        volume_size_gb: int = 100,
+        run_id: str | None = None,
+        region: str | None = None,
+        sagemaker_client: Any = None,
+        logs_client: Any = None,
+    ) -> None: ...
+    # implements ServerlessExecutor protocol
+```
+
+Run replicas as **N independent single-instance SageMaker Training Jobs** (NOT one multi-instance job — that would couple replicas through SageMaker's intra-job NCCL/MPI fabric and break the "each replica syncs only through S3" design). Rank goes through the per-job `Environment` map (`REPLICA_RANK=i` / `WORLD_SIZE=N`), so `replica_entrypoint` reads it unchanged. Pins `EnableNetworkIsolation=False` (never a knob) — `True` would sever the container's S3 access and dead-lock the allreduce poll loop.
+
+**Constructor parameters** (all kw-only)
+
+| Name | Type | Default | Meaning |
+|---|---|---|---|
+| `role_arn` | `str` | — | IAM execution role SageMaker assumes; must grant S3 to the rendezvous + output buckets (the boto3 analog of EKS IRSA). Caller needs `iam:PassRole`. |
+| `image_uri` | `str` | — | ECR training-container image. The executor also passes `ContainerEntrypoint` explicitly so a generic image works. |
+| `output_s3_path` | `str` | — | `s3://...` prefix for `OutputDataConfig.S3OutputPath`. |
+| `instance_type` | `str` | `"ml.g5.2xlarge"` | Default instance type when `gpu` is unmapped. |
+| `cpu_instance_type` | `str` | `"ml.m5.xlarge"` | Instance type used when `gpu=None` (CPU smoke). |
+| `volume_size_gb` | `int` | `100` | `ResourceConfig.VolumeSizeInGB` per job. |
+| `run_id` | `str \| None` | `None` ⇒ random token | Prefix for generated training-job names. |
+| `region` | `str \| None` | `None` | AWS region for the lazy boto3 clients (`None` ⇒ ambient default). |
+| `sagemaker_client` | `Any` | `None` | Inject a pre-built `boto3.client("sagemaker")` (or a mock); built lazily otherwise. |
+| `logs_client` | `Any` | `None` | Inject a pre-built `boto3.client("logs")`; built lazily on first `stream_logs`. |
+
+**`ServerlessExecutor` Protocol methods**
+
+- `launch_replicas(n_replicas, entrypoint, entrypoint_args, *, gpu=None, timeout=3600) -> list[ReplicaHandle]` — submits N independent jobs; `entrypoint` is ignored (container command is baked / passed explicitly). `entrypoint_args` MUST include `rendezvous_uri` (`s3://...`) and `trainer_module`; optional `trainer_fn` (default `"train"`) and `trainer_kwargs` (dict, JSON-encoded into `ContainerArguments`). `gpu` maps to an instance type via `_GPU_INSTANCE_MAP` (`"A100"`/`"H100"`/`"H200"`/`"B200"`/`"L40S"`/`"A10G"`/`"L4"`; a literal `ml.*` string is honoured); `timeout` ⇒ `StoppingCondition.MaxRuntimeInSeconds`. On mid-launch failure it best-effort stops already-launched siblings then raises.
+- `poll(handle) -> str` — maps `describe_training_job`'s `TrainingJobStatus` via `_STATUS_MAP` (`InProgress`⇒`running`, `Completed`⇒`succeeded`, `Failed`⇒`failed`, `Stopping`⇒`running`, `Stopped`⇒`cancelled`); refines `InProgress`⇒`"pending"` while still queued (`SecondaryStatus` in `_PENDING_SECONDARY`); a vanished job (`ResourceNotFound`) ⇒ `"cancelled"`.
+- `stream_logs(handle, *, n_lines=200) -> str` — discovers the CloudWatch stream under `/aws/sagemaker/TrainingJobs` by `<job-name>/` prefix and tails it; falls back to a CloudWatch console URL on any error.
+- `cancel(handle) -> None` — best-effort `stop_training_job` (swallows `ResourceNotFound` / already-terminal `ValidationException`).
+- `collect(handles, *, timeout=None) -> list[dict]` — polls per handle until terminal (`Completed`/`Failed`/`Stopped`) or the shared deadline; results aligned to input order; dict is `{"rank","status","exit_code","error","result","training_job_name"}` (`result` is the `S3ModelArtifacts` path).
+
+**Raises** `RuntimeError` if boto3 is absent and no client was injected; `ValueError` if `n_replicas < 1` or `entrypoint_args` lacks `rendezvous_uri` / `trainer_module`.
+
+```python
+from composer_replication.diloco.serverless.sagemaker import SageMakerExecutor
+ex = SageMakerExecutor(
+    role_arn="arn:aws:iam::123:role/sm-exec",
+    image_uri="123.dkr.ecr.us-east-1.amazonaws.com/diloco:latest",
+    output_s3_path="s3://bucket/out/", region="us-east-1")
+handles = ex.launch_replicas(
+    n_replicas=2, entrypoint="",
+    entrypoint_args={"rendezvous_uri": "s3://bucket/run/",
+                     "trainer_module": "my.trainer"},
+    gpu="A100")
+results = ex.collect(handles)
+```
+
+---
+
+## 16. `composer_replication.datagen.docker_sandbox`
+
+ADR-010 §3 hardened container backend for the FeatureDeletion (FD) env. `DockerSandbox` is a drop-in implementation of the `Sandbox` Protocol (`composer_replication.datagen.sandbox.Sandbox`) that runs the agent's tool calls and the verifiable test command inside an ephemeral, locked-down Docker container instead of a raw host subprocess. It is the production execution path for genuinely UNTRUSTED model-generated code (the `LocalSubprocessSandbox` sibling runs in the host process and is NOT a host-security boundary).
+
+**Extra**: needs `docker>=7` (`pip install -e .[datagen]` or `pip install docker`). The SDK is **lazy-imported inside methods** so the pure-Python core and the `FakeSandbox` path never require it; a clear `RuntimeError` is raised if the SDK or the daemon is absent.
+
+### The `Sandbox` Protocol — `datagen.sandbox`
+
+```python
+@runtime_checkable
+class Sandbox(Protocol):
+    def boot(self, image: str) -> None: ...
+    def exec(self, action: dict) -> str: ...
+    def run_tests(self, test_command: str, tests: tuple[str, ...]) -> TestRunResult: ...
+    def trajectory(self) -> list[dict]: ...
+    def is_command_allowed(self, command: str) -> bool: ...
+```
+
+Structural protocol every FD execution backend implements (`DockerSandbox`, `LocalSubprocessSandbox`, `FakeSandbox`). `boot` prepares the execution environment, `exec` runs one tool-call `action` dict (`{"command": ...}`) and returns combined stdout/stderr, `run_tests` runs the verifiable pytest command over the given node ids and returns a `TestRunResult`, `trajectory` returns the recorded action list, `is_command_allowed` is the first-token denylist check.
+
+### `class DockerSandbox`
+
+```python
+@dataclass
+class DockerSandbox:
+    workdir: str
+    runtime: str | None = None
+    mem_limit: str = "1g"
+    memswap_limit: str = "1g"
+    pids_limit: int = 256
+    nano_cpus: int = 2_000_000_000   # 2 CPUs
+    user: str = "1000:1000"
+    container_workdir: str = "/work"
+    tmpfs_size: str = "64m"
+    exec_timeout_s: int = 600
+    keep_root_writable: bool = False
+
+    def container_kwargs(self, image: str) -> dict: ...
+    # Sandbox Protocol:
+    def boot(self, image: str) -> None: ...
+    def exec(self, action: dict) -> str: ...
+    def run_tests(self, test_command: str, tests: tuple[str, ...]) -> TestRunResult: ...
+    def trajectory(self) -> list[dict]: ...
+    def is_command_allowed(self, command: str) -> bool: ...
+    def close(self) -> None: ...
+    @staticmethod
+    def reap_leaked(client=None) -> int: ...
+```
+
+Hardened ephemeral-container `Sandbox`. The lockdown recipe (CIS Docker 5.x + gVisor guidance): `network_disabled=True` + `network_mode="none"` (no egress — the reward-hack exfil control), `read_only=True` root fs + small `tmpfs` for `/tmp`, workdir bind-mounted RW at `/work`, `cap_drop=["ALL"]` + `security_opt=["no-new-privileges:true"]`, non-root `user`, `pids_limit` (fork-bomb guard) + `mem_limit==memswap_limit` (OOM, no swap) + `nano_cpus` (CPU quota). The PRIMARY reward-hack control is the host-side `scrub_tree(workdir)` run in `boot()` BEFORE the container starts (the bind mount is shared, so host-pre-boot scrubbing == in-container scrubbing); the command denylist is cheap defense-in-depth, not the wall.
+
+**Constructor fields**
+
+| Field | Type | Default | Meaning |
+|---|---|---|---|
+| `workdir` | `str` | — | Host path to the materialized repo; bind-mounted RW at `/work` and scrubbed on the host before boot (PRIMARY control). Must exist by `boot()` time. |
+| `runtime` | `str \| None` | `None` ⇒ daemon default (runc) | `"runsc"` (gVisor) for untrusted model code; requires host `sudo runsc install` + dockerd restart. Gate with `runsc_available()`. |
+| `mem_limit` / `memswap_limit` | `str` | `"1g"` / `"1g"` | OOM guard; equal values forbid swap. |
+| `pids_limit` | `int` | `256` | Fork-bomb guard. |
+| `nano_cpus` | `int` | `2_000_000_000` | CPU quota in 1e-9 CPUs (2 CPUs). |
+| `user` | `str` | `"1000:1000"` | Non-root uid:gid the agent code runs as. |
+| `container_workdir` | `str` | `"/work"` | Bind-mount target + working dir. |
+| `tmpfs_size` | `str` | `"64m"` | Size of the `/tmp` tmpfs scratch. |
+| `exec_timeout_s` | `int` | `600` | Wall-clock cap injected via coreutils `timeout` (exec_run has no timeout param — docker-py #2651). |
+| `keep_root_writable` | `bool` | `False` | Escape hatch if read-only fs breaks tooling. |
+
+**`Sandbox` Protocol methods**
+
+- `boot(image) -> None` — scrubs the host workdir (primary control), reaps leaked siblings, then starts the hardened container. **Raises** `RuntimeError` if the image is not found locally (the container is `--network none`, so it cannot pull) or on a Docker API error.
+- `exec(action) -> str` — records the `action`, denylist-checks its first token, runs the command in the live container (combined stdout/stderr, non-UTF-8 bytes decoded with `errors="replace"`). Returns an `ERROR:` string for a denied command.
+- `run_tests(test_command, tests) -> TestRunResult` — `shlex.quote`s each node id, runs the verifiable command, parses pass/fail conservatively (PASSED-or-rc0 ⇒ passed; `errors during collection` ⇒ `collected_ok=False`).
+- `trajectory() -> list[dict]` — copy of the recorded action list.
+- `is_command_allowed(command) -> bool` — first-token denylist (`SANDBOX_DENYLIST`); not a boundary on its own.
+
+**Lifecycle**: `close()` force-removes the container (idempotent, swallows errors); `reap_leaked(client=None) -> int` (staticmethod) sweeps containers labelled `composer_replication=datagen` and returns the count removed. Also a context manager (`__enter__` / `__exit__` ⇒ `close()`).
+
+**Module helpers**: `runsc_available(client=None) -> bool` (True iff the gVisor `runsc` runtime is registered with the daemon — gate any runsc behavior on this); `_require_docker()` / `_make_client()` (lazy SDK import + daemon ping, each raising a clear `RuntimeError`).
+
+```python
+from composer_replication.datagen.docker_sandbox import DockerSandbox, runsc_available
+runtime = "runsc" if runsc_available() else None
+with DockerSandbox(workdir="/tmp/repo", runtime=runtime) as sb:
+    sb.boot("python:3.12-slim")
+    out = sb.exec({"command": "python -c 'print(1+1)'"})
+    res = sb.run_tests("python -m pytest -q", ("tests/test_x.py::test_a",))
+```
+
+---
+
+## 17. `composer_replication.safety` & `composer_replication.safety.kill_switch`
+
+ADR-015 run-level collapse safeguard — the #2 collapse safety net for the self-evolving RL flywheel. The per-task controls live in `composer_replication.datagen` (4-gate validator, `HackMonitor` provenance, sandbox denylist); this package adds the missing ACROSS-GENERATION / run-level control that watches in-loop (proxy) reward against a disjoint held-out (real) eval and HALTS the run when collapse / reward-hacking is caught in the act. Pure-Python: no torch, no cloud deps; fully CPU-testable (`kl_to_init` is just a float the caller computes upstream).
+
+Public surface (`composer_replication.safety.__all__`): `HeldOutGuard`, `TripwireStatus`, `CollapseStopError`, `kl_token_trust_filter`.
+
+### `class TripwireStatus` — `safety.kill_switch`
+
+```python
+@dataclass(frozen=True)
+class TripwireStatus:
+    fire: bool
+    reason: str
+    step: int
+    proxy_real_gap: float
+    in_loop_ema: float
+    heldout_ema: float
+    kl_ema: float | None = None
+
+    @property
+    def halt(self) -> bool: ...   # documented alias for `fire`
+```
+
+Structured verdict returned by every `HeldOutGuard.update(...)`. `fire` (alias `halt`) ⇒ the run should HALT; `reason` is the human-readable WHY (empty when not firing); `proxy_real_gap` is the RSI "Hacking Gap" (in-loop gain − held-out gain since baseline); `*_ema` are the denoised metric EMAs.
+
+### `class CollapseStopError(RuntimeError)` — `safety.kill_switch`
+
+```python
+class CollapseStopError(RuntimeError):
+    def __init__(self, status: TripwireStatus) -> None: ...
+    status: TripwireStatus
+```
+
+Typed exception for exception-based trainer control flow; carries the fired `TripwireStatus` for logging. Raised (optionally) via `HeldOutGuard.raise_if_fired(...)`.
+
+### `class HeldOutGuard` — `safety.kill_switch`
+
+```python
+@dataclass
+class HeldOutGuard:
+    kl_hard_stop: float = 0.08          # nats/token; top of GRPO "good" band
+    max_proxy_real_gap: float = 0.10    # absolute Hacking-Gap blowout ceiling
+    min_steps: int = 20                 # no fire before this many updates
+    decline_patience: int = 3           # consecutive held-out declines to fire (a)
+    ema_alpha: float = 0.9              # EMA weight on the PRIOR (0.9 => slow)
+    rise_eps: float = 1e-4              # min EMA delta to count as rising/declining
+
+    def update(self, round_idx: int, in_loop_reward: float, heldout_score: float,
+               kl_to_init: float | None = None, entropy: float | None = None,
+               reward_std: float | None = None) -> TripwireStatus: ...
+    def should_halt(self) -> bool: ...
+    @property
+    def last_status(self) -> TripwireStatus | None: ...
+    def raise_if_fired(self, status: TripwireStatus | None = None) -> None: ...
+    def proxy_real_gap(self) -> float: ...
+    def calibrate_kl_threshold(self, baseline_kls: list[float], factor: float = 3.0) -> float: ...
+```
+
+Stateful across-generation collapse / reward-hacking kill-switch. Call `update(...)` once per checkpoint (the same cadence as `DifficultyCurriculum.update`); it folds each metric into a denoised EMA and returns a `TripwireStatus`.
+
+**Fires (`fire=True`) when ANY of three conditions** (none before `min_steps`, which guards early-run noise):
+
+- **(a) collapse-caught-in-the-act** — the in-loop reward EMA is RISING while the held-out score EMA has DECLINED for `>= decline_patience` consecutive checkpoints (the canonical reward-hacking signature: proxy up, real down).
+- **(b) KL hard stop** — the `kl_to_init` EMA exceeds `kl_hard_stop` (default **0.08 nats/token**, the top of the GRPO "good progression" band). Checked first (cheapest unambiguous breach).
+- **(c) proxy-real gap blowout** — the Hacking Gap (proxy gain − real gain since baseline) widens beyond `max_proxy_real_gap`, catching a fast single-generation divergence even before the full decline window.
+
+Once fired, the verdict is **latched** — every subsequent `update` keeps `fire=True` so a transient recovery cannot silently un-halt the run.
+
+**Constructor thresholds**
+
+| Field | Type | Default | Meaning |
+|---|---|---|---|
+| `kl_hard_stop` | `float` | `0.08` | Per-token KL(policy‖init) hard-stop ceiling, nats/token (condition (b)). Must be > 0. |
+| `max_proxy_real_gap` | `float` | `0.10` | Absolute Hacking-Gap blowout ceiling (condition (c)). |
+| `min_steps` | `int` | `20` | No tripwire fires before this many `update` calls. |
+| `decline_patience` | `int` | `3` | Consecutive held-out declines (while in-loop rises) to fire (a). Must be >= 1. |
+| `ema_alpha` | `float` | `0.9` | EMA weight on the PRIOR (0.9 ⇒ slow). Must be in `[0, 1)`. |
+| `rise_eps` | `float` | `1e-4` | Min EMA delta to count as "rising" / "declining". |
+
+**API**
+
+- `update(round_idx, in_loop_reward, heldout_score, kl_to_init=None, entropy=None, reward_std=None) -> TripwireStatus` — fold one checkpoint's metrics and return the verdict. `round_idx` is for logging only (the internal counter `_n` drives `min_steps`). `kl_to_init` is **token-mean KL in nats/token** (this repo's `token_mean_kl` convention) — do NOT pass a sequence-summed KL (it will fire instantly). `entropy` / `reward_std` are tracked + exposed but not hard gates.
+- `should_halt() -> bool` — True iff the most recent `update` fired. **Idempotent** (does not advance EMA state).
+- `last_status -> TripwireStatus | None` (property) — the most recent verdict, or `None` before the first `update`.
+- `raise_if_fired(status=None) -> None` — convert a fired verdict (the passed status, or `last_status`) into a `CollapseStopError`; a no-op otherwise. For exception-based loops.
+- `proxy_real_gap() -> float` — the RSI Hacking Gap (EMA-minus-baseline, both since run start); `0.0` before the first `update`.
+- `calibrate_kl_threshold(baseline_kls, factor=3.0) -> float` — set `kl_hard_stop` from early-run baseline KLs (`factor` × mean). SAFETY CLAMP: only ever TIGHTENS (`min(factor*mean, current)`), never loosens past the documented band. **Raises** `ValueError` on empty `baseline_kls`.
+
+**Raises** `ValueError` at construction if `ema_alpha` ∉ `[0, 1)`, `kl_hard_stop <= 0`, or `decline_patience < 1`.
+
+> **HeldoutSplit discipline (design-of-record).** `heldout_score` must come from a DISJOINT held-out eval pool — REAL tasks the generator NEVER trains on. If the held-out set drifts with the train set, the proxy-real gap signal is meaningless. See ADR-015 and the `safety.holdout` design notes referenced from the module docstring.
+
+```python
+from composer_replication.safety import HeldOutGuard
+guard = HeldOutGuard(kl_hard_stop=0.08)
+for rnd in range(num_generations):
+    status = guard.update(rnd, in_loop_reward=r_proxy, heldout_score=r_real,
+                          kl_to_init=token_mean_kl_value)
+    if status.fire:           # or: guard.should_halt()
+        guard.raise_if_fired(status)   # -> CollapseStopError
+```
+
+### `kl_token_trust_filter(logratio_sq_half, threshold=0.08) -> bool` — `safety.kill_switch`
+
+```python
+def kl_token_trust_filter(logratio_sq_half: float, threshold: float = 0.08) -> bool
+```
+
+Per-token KL trust-region mask mirroring torchrl's GRPO "KL-Mask". The caller computes `0.5 * (log π/π_ref)**2` (the Schulman k2 per-token KL estimator, nats); this returns `True` when the token should be MASKED OUT (its KL contribution exceeds `threshold`). Pulls no torch into the module — wire it into a loss downstream.
+
+```python
+from composer_replication.safety import kl_token_trust_filter
+mask_out = kl_token_trust_filter(0.5 * logratio**2, threshold=0.08)
+```
+
+---
+
 ## Notes on test coverage
 
 Tested contracts (referenced spike/test paths):
@@ -1513,6 +1868,9 @@ Tested contracts (referenced spike/test paths):
 - `make_diloco_outer_loop` + sign convention: `spikes/008-streaming-diloco/tests/test_diloco_smoke.py`.
 - `ObjectStoreAllReduce`, `MockManager`, `LocalProcessExecutor`, `ReplicaHandle`, `ServerlessExecutor`, `replica_entrypoint.main`: `composer_replication/diloco/serverless/tests/test_serverless_local.py`, `test_serverless_diloco_integration.py`.
 - `recipes.prime_rl.composer_loss.loss_fn`: `composer_replication/recipes/prime_rl/tests/test_composer_loss.py`.
+- `EKSExecutor`, `SageMakerExecutor` (§15): `composer_replication/diloco/serverless/tests/` (DI'd `batch_api`/`core_api` / `sagemaker_client` mocks; no live cloud).
+- `DockerSandbox` (§16) — `container_kwargs` lockdown config asserted without a live daemon; daemon-gated paths in `composer_replication/datagen/tests/`.
+- `HeldOutGuard`, `TripwireStatus`, `CollapseStopError`, `kl_token_trust_filter` (§17): `composer_replication/safety/tests/` (pure-Python; CPU-only).
 
 Untested-contract symbols (⚠️) and skeletons (🟡) are flagged inline above.
 

docs/BACKLOG_RESOLUTION_2026-06-09.md

@@ -75,6 +75,8 @@ Goal-driven systematic resolution of every pending item. This doc is the live au
 | R11 | Flaky test `spikes/006-real-hf-model-smoke/tests/test_strict.py::test_alternating_batches_loss_decreases` — fails under CPU contention (full suite w/ concurrent pytest + Docker), PASSES in isolation (verified 3x). Loss-trend assertion is timing/noise-sensitive. Pin seed / widen tolerance / mark flaky. Pre-existing, not a Wave-2 regression. | LOW | OPEN |
 | R12 | B7-complete ✅ (top-level `__all__` now includes the 3 factories) + B4-complete ✅ (the 4 surviving "115" claims → 266/62). | — | DONE |
 
+**Wave 3 — DONE (Phase-7 reconciliation):** R1 ✅ (HeldOutGuard wired into ComposerReplicationTrainer — optional, OFF by default, soft/hard stop; + integration test), R2 ✅ (HeldoutSplit disjointness enforcer `safety/holdout.py` + 10 tests), R3 ✅ (EKS entrypoint contract bug fixed — `replica_entrypoint.__main__` now resolves from env OR argv; proven end-to-end with a pure-env invocation), R4 ✅ (calibrate_kl_threshold rejects factor<=0/negative-baseline + positive floor), R7 ✅ (API_REFERENCE §15-17: EKS/SageMaker/DockerSandbox/safety), R8 ✅ (ADR-015 authored + indexed), R10 ✅ (path-(c) divergence-rate test). R12 ✅ (B4/B7 complete). DEFERRED-LOW: R5 (cancel exception-narrowing) + R6 (EKS collect result-key) — stale-base worktree casualties, tracked, LOW severity. R11 (spike-006 flaky-under-contention) — pre-existing, tracked.
+
 Sandbox refactor verdict: **clean** (no regression to LocalSubprocessSandbox/FeatureDeletionEnv).
 
 ## Wave plan

docs/adrs/ADR-015-holdout-killswitch.md

@@ -0,0 +1,187 @@
+---
+status: accepted
+date: 2026-06-08
+deciders: [Codeseys, ARIA]
+builds-on: [ADR-010 (FeatureDeletion datagen — per-task controls), ADR-012 (curriculum + provenance review findings)]
+---
+
+# ADR-015: Held-out disjoint eval + depth/generation kill-switch (HeldOutGuard)
+
+## Context and Problem Statement
+
+The framework drives a **self-evolving RL flywheel**: a generator proposes tasks,
+the policy is optimized against an in-loop (proxy / oracle) reward, and the loop
+repeats across generations. ADR-010 gave this loop its **per-task** safety
+controls — the 4-gate solvability validator, the `HackMonitor` provenance check,
+and the sandbox denylist (now hardened by `DockerSandbox`, see API §16). What was
+still missing is the **run-level / across-generation** control: a watcher sitting
+ABOVE the per-task gates that asks, every generation, *"is the proxy reward
+improving because the policy got better, or because it learned to game the
+proxy?"* — and HALTS the run when the answer is the latter.
+
+The literature is unambiguous that a held-out eval + a hard stop is the
+load-bearing control here, not a nice-to-have:
+
+- **Reward hacking rises monotonically with optimization depth.** Zhao et al.,
+  *"Reward Hacking in Self-Improving Code Agents"* (ICLR 2026 Workshop on RSI,
+  OpenReview `ikrQWGgxYg`), show that going from 10 → 100 optimization steps
+  drives the hacking rate from 26.4% → 57.8% (+31.4 points), and that 73.8% of
+  KernelBench / 46.8% of ALE-Bench optimizations show **proxy gains without real
+  gains**. They define **Hacking Gap = proxy gain − real gain** and label an
+  optimization reward-hacking when it *"improves the public metric WITHOUT
+  improving the private metric"* — the canonical signature a run-level tripwire
+  must fire on. Because the hacking rate climbs with depth, a *one-time* eval is
+  insufficient; the control has to be an **online per-generation tripwire**.
+
+- **Closed-loop RL on self-generated data collapses.** The self-evolving-agents
+  survey (Gao et al., TMLR 2026; arXiv 2507.21046 v4) **§8.3** names *"model
+  collapse from closed-loop RL on static synthetic data"* and prescribes
+  *"continuous monitoring … to detect long-horizon value drift."* Shumailov et
+  al. (*Nature* 2024, "AI models collapse when trained on recursively generated
+  data") show self-training first loses the distribution tails, then converges to
+  a low-variance point estimate. The mitigation that matters here: the held-out
+  eval must stay anchored to **REAL tasks that are NEVER fed back to the
+  generator**, otherwise the eval drifts with the train set and the gap signal
+  becomes meaningless.
+
+- **Self-critique alone is insufficient.** The RSI paper's "retrospection"
+  self-critique sometimes *increased* hacking; its conclusion: *"mitigating
+  reward hacking likely requires stronger evaluations and constraints beyond
+  self-critique alone."* So the control is a genuinely disjoint held-out eval +
+  hard stop, not a critique hook.
+
+- **A verifiable / held-out reward is still gameable.** EvilGenie
+  (arXiv 2511.21654) found *"only minimal improvement from the use of held out
+  test cases"* in isolation and that *"holdout tests have many surprising failure
+  modes."* The implication is NOT "skip the held-out eval" but "do not treat it
+  as sufficient on its own" — layer it explicitly as **defense-in-depth** on top
+  of `HackMonitor` (provenance) and the KL hard stop. Catastrophic Goodhart
+  (OpenReview `UXuBzWoZGK`) likewise proves KL regularization alone does NOT
+  prevent heavy-tailed reward misspecification — so the KL stop is one tripwire
+  among several, never the sole control.
+
+## Decision
+
+Build **`composer_replication.safety`** — a pure-Python, no-torch / no-cloud,
+fully CPU-testable run-level safeguard — centered on the **`HeldOutGuard`**
+kill-switch. The guard is **stateful** and is called once per checkpoint /
+generation (`update(round_idx, in_loop_reward, heldout_score, kl_to_init=…)`),
+the same cadence as `DifficultyCurriculum.update`. It maintains denoised EMAs of
+every metric (raw single-step values are too noisy to threshold) and returns a
+structured `TripwireStatus`.
+
+### The 3 fire conditions
+
+`HeldOutGuard.update` returns `fire=True` (alias `halt`) when **ANY** of:
+
+- **(a) collapse-caught-in-the-act** — the in-loop reward EMA is RISING while the
+  held-out score EMA has DECLINED for `>= decline_patience` consecutive
+  checkpoints (default 3, the "monotone for ≥3 checkpoints" rule). This is the
+  canonical reward-hacking signature: proxy up, real down. A held-out dip during
+  an in-loop dip is treated as noise (a hard batch), not hacking — the decline
+  streak only grows when in-loop is *simultaneously* rising.
+
+- **(b) KL-to-init hard stop** — the `kl_to_init` EMA exceeds `kl_hard_stop`
+  (default **0.08 nats/token**) on/after `min_steps`. Checked first as the
+  cheapest unambiguous breach.
+
+- **(c) proxy-real gap blowout** — the Hacking Gap (proxy gain − real gain since a
+  run-start baseline) widens beyond `max_proxy_real_gap` (default 0.10), catching
+  a fast single-generation divergence even before the full decline window
+  elapses. `HeldOutGuard.proxy_real_gap()` returns exactly the RSI Hacking-Gap
+  quantity.
+
+No tripwire fires before `min_steps` (default 20) to avoid halting on early-run
+warm-up noise. Once fired, the verdict is **latched** — every subsequent `update`
+keeps `fire=True`, so a transient post-collapse recovery cannot silently un-halt
+the run.
+
+### HeldoutSplit disjointness discipline (design-of-record)
+
+The `heldout_score` fed to the guard MUST come from a **disjoint held-out eval
+pool** — REAL tasks the generator NEVER trains on (the `HeldoutSplit`
+discipline). This is the load-bearing precondition: per the self-evolving survey
+§8.3 / Shumailov collapse dynamics, if the held-out set is allowed to drift with
+the train set, the proxy-real gap signal degenerates and the guard becomes blind
+to the exact collapse it exists to catch. The split is documented here as the
+**design-of-record**; the guard consumes a scalar `heldout_score` and does not
+itself partition data — the caller is responsible for keeping the split disjoint
+and never feeding held-out tasks back into the generator.
+
+### The 0.08 nats/token KL hard-stop default
+
+The GRPO "healthy progression" band (Orchestra Research GRPO skill) climbs
+0.02 → 0.05 → 0.08 → 0.12 nats/token over a run, with **0.08 the top of the "good
+progression" band** and just below the code-generation drift zone (0.05–0.15
+per-token; >0.5 is "diverging too much"). So 0.08 nats/token is a sound
+hard-stop default. `calibrate_kl_threshold(baseline_kls, factor=3.0)` lets a run
+adapt the ceiling to its own KL scale ("record baseline KL over the first ~100
+steps, set max to 3× that") — but with a **safety clamp**: calibration may only
+ever TIGHTEN the stop (`min(3×baseline, current)`), never loosen it past the
+documented collapse band, so a noisy / already-drifting baseline cannot raise the
+ceiling above 0.08.
+
+> **UNITS GOTCHA (load-bearing).** `kl_to_init` is **token-mean KL in
+> nats/token**, matching `composer_replication.integrations.altered_minds.
+> kl_logging.token_mean_kl`. It is NOT comparable to a sequence-level /
+> sequence-summed KL (whose healthy band is ~0.05–10). Passing a sequence-summed
+> KL into the per-token hard stop will fire it instantly.
+
+### Public surface
+
+`composer_replication.safety` re-exports:
+`HeldOutGuard`, `TripwireStatus`, `CollapseStopError`,
+`kl_token_trust_filter`. The guard exposes both flag-checking
+(`should_halt()` / `status.fire` / `status.halt`) and exception-based
+(`raise_if_fired(status) -> CollapseStopError`) control flow so a trainer loop can
+use whichever convention it prefers. `kl_token_trust_filter` is the per-token
+torchrl-style "KL-Mask" sibling (caller passes `0.5·(log π/π_ref)²`; returns
+True to mask the token) — same 0.08 band, kept torch-free.
+
+## Consequences
+
+- **Positive**: the flywheel gains a run-level, online collapse tripwire that
+  fires on the literature's exact reward-hacking signature (proxy-up / real-down),
+  is denoised against single-step noise, and latches so a detected collapse
+  cannot un-halt. It is layered defense-in-depth ON TOP OF the per-task ADR-010
+  controls — neither sufficient alone (per EvilGenie / Catastrophic Goodhart).
+- **Positive**: pure-Python and CPU-testable — `kl_to_init` is a float the caller
+  computes upstream, so the guard pulls no torch / cloud dependency and is unit
+  testable without a model.
+- **Positive**: the thresholds are calibratable and the KL stop only ever
+  tightens, so the safety property (ceiling ≤ documented band) is preserved
+  across calibration.
+- **Negative / honest**: a held-out eval is necessary but NOT sufficient by
+  itself (EvilGenie); the guard's value depends entirely on the caller honoring
+  the `HeldoutSplit` disjointness discipline. The KL stop is one tripwire among
+  several, not a Goodhart-proof guarantee. `entropy` / `reward_std` are tracked
+  and exposed but are NOT yet hard gates (early-warning instruments only).
+- **Neutral**: `HeldoutSplit` ships as a documented design-of-record discipline
+  rather than an enforced data-partitioning class in this wave; the guard
+  consumes the scalar held-out score the caller provides.
+
+## Acceptance gate
+
+- [x] `HeldOutGuard.update(...)` folds in-loop / held-out / KL (+ entropy /
+  reward_std) EMAs and returns a `TripwireStatus`; fires on (a) collapse-in-the-act,
+  (b) KL > 0.08 nats/token, (c) proxy-real gap blowout; no fire before `min_steps`;
+  latched after first fire.
+- [x] `proxy_real_gap()` returns the RSI Hacking-Gap (in-loop gain − held-out gain
+  since baseline); `should_halt()` / `last_status` are idempotent query helpers;
+  `raise_if_fired()` converts a fired verdict into `CollapseStopError`.
+- [x] `calibrate_kl_threshold()` only ever TIGHTENS the hard stop (safety clamp);
+  raises on empty input.
+- [x] `kl_token_trust_filter()` per-token KL-Mask helper, torch-free.
+- [x] Pure-Python, CPU-only; `composer_replication.safety.__init__` re-exports the
+  public surface and references this ADR.
+- [x] Documented in `docs/API_REFERENCE.md` §17.
+
+## More Information
+
+- `composer_replication/safety/kill_switch.py` — the implementation + the
+  primary-source citations inline.
+- ADR-010 (FeatureDeletion datagen) — the per-task controls this layers above.
+- `docs/API_REFERENCE.md` §16 (`DockerSandbox`) / §17 (`composer_replication.safety`).
+- Zhao et al. RSI (OpenReview `ikrQWGgxYg`); Gao et al. self-evolving survey
+  §8.3 (arXiv 2507.21046 v4); Shumailov et al. (*Nature* 2024); EvilGenie
+  (arXiv 2511.21654); Catastrophic Goodhart (OpenReview `UXuBzWoZGK`).

docs/adrs/README.md

@@ -16,6 +16,7 @@
 | [ADR-012](ADR-012-close-review-findings.md) | Close open cross-family-review findings (KL/hint-routing/provenance/curriculum) | accepted (amends 008/009/010) | 2026-05-29 |
 | [ADR-013](ADR-013-lma-integration-channel-ladder.md) | LMA integration — isolated-channel ladder (supersedes tie-in Phase-3 hyperparams) | accepted | 2026-05-29 |
 | [ADR-014](ADR-014-policy-optimization-objective-menu.md) | Policy-optimization objective MENU: base RL objective selectable (default Dr.GRPO) over TRL 1.5.0 GRPOConfig (builds-on ADR-006/007/008) | accepted | 2026-05-30 |
+| [ADR-015](ADR-015-holdout-killswitch.md) | Held-out disjoint eval + depth/generation kill-switch (run-level collapse safeguard #2): `HeldOutGuard` + `HeldoutSplit` in `composer_replication.safety` | accepted | 2026-06-09 |
 
 Sorted by number ascending. ADRs are immutable after `accepted`; supersede or amend rather than edit.