feat(train): --replay / --replay-ratio passthrough (v0.71.36 Part E)

Thin CLI overrides for data.replay / data.replay_ratio, mirroring the
--reward-hack-mitigation style.

_apply_replay_overrides rebuilds the config rather than mutating it in
place, and that IS the mechanism: rebuilding re-runs every cross-validator,
so `--replay` on task='dpo' hits _validate_replay_compat exactly as a YAML
value would. Mutating in place would let CLI flags bypass every gate --
mutation-verified: the in-place version fails 3 named tests. It also leaves
the caller's config untouched.

Declared the flags as plain str/float with a None default, matching this
file's existing convention (name/resume/annex_xi). train.py has
`from __future__ import annotations` but never imports Optional, and Typer
must resolve annotations at runtime -- Optional[str] raised NameError at
--help time, caught immediately by the help test.

src/soup_cli/commands/train.py

@@ -137,6 +137,25 @@ def _hardware_fit_preflight(cfg, gpu_info, *, allow_oom_attempt: bool) -> None:
         raise typer.Exit(1)
 
 
+def _apply_replay_overrides(cfg, *, replay, replay_ratio):
+    """Apply ``--replay`` / ``--replay-ratio``, then RE-VALIDATE.
+
+    Re-validation is the point: a CLI override must clear the same
+    cross-validators as YAML, or ``--replay`` on ``task='dpo'`` would slip
+    past ``_validate_replay_compat``. Rebuilding the model (rather than
+    mutating in place) is what re-runs them, and leaves the caller's config
+    untouched.
+    """
+    if replay is None and replay_ratio is None:
+        return cfg
+    payload = cfg.model_dump()
+    if replay is not None:
+        payload["data"]["replay"] = replay
+    if replay_ratio is not None:
+        payload["data"]["replay_ratio"] = replay_ratio
+    return type(cfg)(**payload)
+
+
 def train(
     config: str = typer.Option(
         "soup.yaml",
@@ -295,6 +314,22 @@ def train(
             "--reward-hack-detector (or training.reward_hack_detector). (v0.71.26)"
         ),
     ),
+    replay: str = typer.Option(
+        None, "--replay",
+        help=(
+            "Old dataset to interleave as continual-learning rehearsal, so "
+            "training on the new task does not erase the previous one. "
+            "sft/pretrain only; incompatible with packing/multipack. "
+            "Overrides data.replay. (v0.71.36)"
+        ),
+    ),
+    replay_ratio: float = typer.Option(
+        None, "--replay-ratio",
+        help=(
+            "Fraction of the FINAL mixed train set that is replay rows "
+            "(default 0.1). Overrides data.replay_ratio. (v0.71.36)"
+        ),
+    ),
     reward_hack_mitigation: str = typer.Option(
         None,
         "--reward-hack-mitigation",
@@ -466,6 +501,15 @@ def train(
     console.print(f"[dim]Loading config from {config_path}...[/]")
     cfg = load_config(config_path)
 
+    # --- v0.71.36 replay passthrough ---
+    try:
+        cfg = _apply_replay_overrides(
+            cfg, replay=replay, replay_ratio=replay_ratio
+        )
+    except Exception as exc:  # noqa: BLE001 — pydantic ValidationError et al.
+        console.print(f"[red]{markup_escape(str(exc))}[/]")
+        raise typer.Exit(code=2) from exc
+
     # --- RA-DIT generator-stage auto-link (v0.71.10 #200) ---
     # When a generator stage has no retriever model set, splice in the latest
     # RA-DIT retriever output from the Registry. A manual value always wins.

tests/test_v07136.py

@@ -2415,3 +2415,83 @@ class TestLoaderReplay:
         assert source.count("def _finalize(") == 1
         # local + remote + hf paths
         assert source.count("_finalize(") >= 4
+
+
+class TestTrainReplayFlags:
+    def test_help_lists_replay(self):
+        from typer.testing import CliRunner
+
+        from soup_cli.cli import app
+
+        res = CliRunner(env={"COLUMNS": "200"}).invoke(app, ["train", "--help"])
+        assert res.exit_code == 0, (res.output, repr(res.exception))
+        cleaned = _clean(res.output)
+        assert "--replay" in cleaned
+        assert "--replay-ratio" in cleaned
+
+    def _cfg(self, yaml_str=None):
+        from soup_cli.config.loader import load_config_from_string
+
+        return load_config_from_string(yaml_str or _PLAIN_YAML)
+
+    def test_apply_replay_overrides(self):
+        from soup_cli.commands.train import _apply_replay_overrides
+
+        out = _apply_replay_overrides(
+            self._cfg(), replay="old.jsonl", replay_ratio=0.25
+        )
+        assert out.data.replay == "old.jsonl"
+        assert out.data.replay_ratio == 0.25
+
+    def test_no_flags_is_identity(self):
+        from soup_cli.commands.train import _apply_replay_overrides
+
+        cfg = self._cfg()
+        out = _apply_replay_overrides(cfg, replay=None, replay_ratio=None)
+        assert out is cfg
+        assert out.data.replay is None
+
+    def test_override_is_revalidated_against_the_gate(self):
+        """--replay on task=dpo must hit _validate_replay_compat.
+
+        A CLI override that skipped re-validation would slip past every
+        cross-validator that YAML has to satisfy.
+        """
+        from soup_cli.commands.train import _apply_replay_overrides
+
+        dpo = self._cfg(
+            "base: m\ntask: dpo\ndata:\n  train: t.jsonl\n  format: dpo\n"
+            "training:\n  epochs: 1\n"
+        )
+        with pytest.raises(Exception, match="replay"):
+            _apply_replay_overrides(dpo, replay="old.jsonl", replay_ratio=None)
+
+    def test_ratio_without_replay_flag_rejected(self):
+        from soup_cli.commands.train import _apply_replay_overrides
+
+        with pytest.raises(Exception, match="replay"):
+            _apply_replay_overrides(self._cfg(), replay=None, replay_ratio=0.3)
+
+    def test_bad_ratio_rejected(self):
+        from soup_cli.commands.train import _apply_replay_overrides
+
+        with pytest.raises(Exception):
+            _apply_replay_overrides(
+                self._cfg(), replay="old.jsonl", replay_ratio=0.9
+            )
+
+    def test_yaml_value_preserved_when_flag_absent(self):
+        """--replay-ratio alone must not clobber a YAML data.replay."""
+        from soup_cli.commands.train import _apply_replay_overrides
+
+        cfg = self._cfg(_REPLAY_YAML)
+        out = _apply_replay_overrides(cfg, replay=None, replay_ratio=0.3)
+        assert out.data.replay == "old.jsonl"
+        assert out.data.replay_ratio == 0.3
+
+    def test_override_does_not_mutate_the_input_config(self):
+        from soup_cli.commands.train import _apply_replay_overrides
+
+        cfg = self._cfg()
+        _apply_replay_overrides(cfg, replay="old.jsonl", replay_ratio=0.2)
+        assert cfg.data.replay is None, "input config must not be mutated"