fix(loader): replay rows bypassed vision/audio traversal protection (HIGH)

ECC code-review found a real bug and I reproduced it before fixing.

load_dataset runs _validate_vision_images / _validate_audio_files on the
primary dataset -- they exist to reject `{"image": "/etc/passwd"}` -- but
the replay file is loaded by _load_replay_rows, which did neither. A
genuinely llava-shaped replay row keeps its `image` value through
format_to_messages, and nothing gated data.replay on modality, so a
traversal path from the replay file reached PIL.Image.open in the trainer.

Verified rather than assumed: a chatml+image row is detected as chatml and
its image key is dropped (my first probe found 0 escapes, which is why the
shape matters), but a real llava row -> detect_format='llava' -> the
traversal path survives conversion intact. Both the vision and audio cases
now have failing-first tests; the vision guard is mutation-verified.

Fix: the replay file gets its own containment pass, resolving media against
the REPLAY file's directory (the old dataset's images live with the old
dataset) unless an explicit image_dir/audio_dir is configured.

Also from the same review:
- MEDIUM: `canary insert` wrote the poisoned dataset BEFORE the manifest, so
  a manifest failure left canaried data on disk with nothing to identify the
  secrets in it. Manifest is now written first; if the data write then
  fails, the error says the manifest describes canaries that were never
  inserted rather than leaving it looking authoritative.
- LOW: --threshold help said "MinHash similarity" but --semantic reuses it
  for embedding cosine.
- LOW: added --replay-seed, which had no CLI override unlike its two sibling
  flags.

260 new-file tests + 337 loader/vision/audio regression tests green.

src/soup_cli/commands/data.py

@@ -295,7 +295,8 @@ def dedup(
     ),
     threshold: float = typer.Option(
         0.8, "--threshold",
-        help="MinHash similarity threshold (0.0-1.0)",
+        help="Similarity threshold (0.0-1.0): MinHash Jaccard by default, "
+             "embedding cosine under --semantic.",
     ),
     field: str = typer.Option(
         None, "--field", "-f",

src/soup_cli/commands/data_canary.py

@@ -92,14 +92,31 @@ def insert(
         raise typer.Exit(1)
 
     mixed = list(rows) + canary_rows(canaries)
-    atomic_write_text(
-        "\n".join(json.dumps(row) for row in mixed) + "\n", output
-    )
+    # Manifest FIRST. If the dataset were written first and the manifest
+    # then failed (disk full, permissions), a canary-poisoned dataset would
+    # survive on disk with nothing left to identify the secrets in it — a
+    # user who missed the error and trained on it could never audit what
+    # was inserted. Failing before the data is written leaves no artifact.
     try:
         write_manifest(canaries, manifest)
     except (ValueError, OSError) as exc:
         console.print(f"[red]{escape(str(exc))}[/]")
         raise typer.Exit(1)
+    try:
+        atomic_write_text(
+            "\n".join(json.dumps(row) for row in mixed) + "\n", output
+        )
+    except (ValueError, OSError) as exc:
+        # The manifest now describes canaries that are in no dataset. Say so
+        # rather than leaving a manifest that looks authoritative.
+        console.print(
+            f"[red]Could not write {escape(str(output))}: "
+            f"{escape(str(exc))}[/]\n"
+            f"[yellow]{escape(str(manifest))} was already written and now "
+            "describes canaries that were NOT inserted — delete it or re-run."
+            "[/]"
+        )
+        raise typer.Exit(1)
 
     console.print(
         f"[green]Inserted {len(canaries)} canaries:[/] "

src/soup_cli/commands/train.py

@@ -137,8 +137,8 @@ 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.
+def _apply_replay_overrides(cfg, *, replay, replay_ratio, replay_seed=None):
+    """Apply the ``--replay*`` flags, 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
@@ -146,13 +146,15 @@ def _apply_replay_overrides(cfg, *, replay, replay_ratio):
     mutating in place) is what re-runs them, and leaves the caller's config
     untouched.
     """
-    if replay is None and replay_ratio is None:
+    if replay is None and replay_ratio is None and replay_seed 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
+    if replay_seed is not None:
+        payload["data"]["replay_seed"] = replay_seed
     return type(cfg)(**payload)
 
 
@@ -330,6 +332,13 @@ def train(
             "(default 0.1). Overrides data.replay_ratio. (v0.71.36)"
         ),
     ),
+    replay_seed: int = typer.Option(
+        None, "--replay-seed",
+        help=(
+            "Seed for the replay sample + interleave. Overrides "
+            "data.replay_seed. (v0.71.36)"
+        ),
+    ),
     reward_hack_mitigation: str = typer.Option(
         None,
         "--reward-hack-mitigation",
@@ -504,7 +513,10 @@ def train(
     # --- v0.71.36 replay passthrough ---
     try:
         cfg = _apply_replay_overrides(
-            cfg, replay=replay, replay_ratio=replay_ratio
+            cfg,
+            replay=replay,
+            replay_ratio=replay_ratio,
+            replay_seed=replay_seed,
         )
     except Exception as exc:  # noqa: BLE001 — pydantic ValidationError et al.
         console.print(f"[red]{markup_escape(str(exc))}[/]")

src/soup_cli/data/loader.py

@@ -116,6 +116,15 @@ def _load_replay_rows(data_config: DataConfig) -> list[dict]:
 
     The old dataset may be alpaca while the new one is sharegpt, so the
     replay file cannot inherit ``data_config.format``.
+
+    It also gets its OWN media-containment pass. ``load_dataset`` runs
+    :func:`_validate_vision_images` / :func:`_validate_audio_files` on the
+    primary dataset, and the replay file is loaded here rather than there —
+    so without this, a llava-shaped replay row's ``image`` value survives
+    ``format_to_messages`` untouched and a traversal path would reach
+    ``PIL.Image.open`` in the trainer. Media resolve against the REPLAY
+    file's own directory unless an explicit dir is configured: the old
+    dataset's images live with the old dataset.
     """
     replay_path = Path(data_config.replay)
     if not is_under_cwd(replay_path):
@@ -129,7 +138,23 @@ def _load_replay_rows(data_config: DataConfig) -> list[dict]:
     raw = load_raw_data(replay_path)
     fmt = detect_format(raw)
     rows = [format_to_messages(row, fmt) for row in raw]
-    return [row for row in rows if row is not None]
+    rows = [row for row in rows if row is not None]
+
+    if is_vision_format(fmt):
+        image_dir = (
+            Path(data_config.image_dir)
+            if data_config.image_dir
+            else replay_path.parent
+        )
+        rows = _validate_vision_images(rows, image_dir)
+    if is_audio_format(fmt):
+        audio_dir = (
+            Path(data_config.audio_dir)
+            if data_config.audio_dir
+            else replay_path.parent
+        )
+        rows = _validate_audio_files(rows, audio_dir)
+    return rows
 
 
 def _finalize(

tests/test_v07136.py

@@ -1718,6 +1718,63 @@ class TestDataCanaryCli:
         out = _clean(res.output).lower()
         assert "commit" in out or "secret" in out
 
+    def test_manifest_is_written_before_the_poisoned_dataset(
+        self, tmp_path, monkeypatch
+    ):
+        """Order matters: a dataset with no manifest is unauditable.
+
+        If the data were written first and the manifest then failed, a
+        canary-poisoned dataset would survive on disk with nothing left to
+        say what was inserted into it.
+        """
+        from pathlib import Path
+
+        from typer.testing import CliRunner
+
+        from soup_cli.cli import app
+        from soup_cli.commands import data_canary as cmd
+
+        monkeypatch.chdir(tmp_path)
+        src = self._dataset(tmp_path, count=3)
+
+        def _boom(canaries, path):
+            raise OSError("disk full")
+
+        monkeypatch.setattr(cmd, "write_manifest", _boom)
+        res = CliRunner().invoke(
+            app,
+            ["data", "canary", "insert", str(src), "-o", "out.jsonl",
+             "--manifest", "m.json"],
+        )
+        assert res.exit_code == 1
+        assert not Path("out.jsonl").exists(), (
+            "a canary-poisoned dataset must not survive a manifest failure"
+        )
+
+    def test_data_write_failure_warns_the_manifest_is_now_stale(
+        self, tmp_path, monkeypatch
+    ):
+        from typer.testing import CliRunner
+
+        from soup_cli.cli import app
+        from soup_cli.commands import data_canary as cmd
+
+        monkeypatch.chdir(tmp_path)
+        src = self._dataset(tmp_path, count=3)
+
+        def _boom(text, path):
+            raise OSError("disk full")
+
+        monkeypatch.setattr(cmd, "atomic_write_text", _boom)
+        res = CliRunner().invoke(
+            app,
+            ["data", "canary", "insert", str(src), "-o", "out.jsonl",
+             "--manifest", "m.json"],
+        )
+        assert res.exit_code == 1
+        out = _clean(res.output).lower()
+        assert "not inserted" in out or "re-run" in out
+
     def test_insert_output_outside_cwd_rejected(self, tmp_path, monkeypatch):
         from typer.testing import CliRunner
 
@@ -2404,6 +2461,86 @@ class TestLoaderReplay:
         assert len(old) == 3
         assert len(set(old)) == 3, "rows must not be repeated"
 
+    def _llava(self, tag, image, count):
+        return [
+            {
+                "image": image,
+                "conversations": [
+                    {"from": "human", "value": "<image>\ndescribe"},
+                    {"from": "gpt", "value": f"{tag}{i}"},
+                ],
+            }
+            for i in range(count)
+        ]
+
+    def test_replay_vision_rows_get_traversal_protection(
+        self, tmp_path, monkeypatch
+    ):
+        """The replay file must get the SAME image containment as the main one.
+
+        load_dataset runs _validate_vision_images on the primary dataset
+        (it exists to reject `{"image": "/etc/passwd"}`), but the replay
+        file is loaded by its own path. A genuinely llava-shaped replay row
+        keeps its `image` value through format_to_messages, so without a
+        validation pass a traversal path would reach PIL.Image.open.
+        """
+        from soup_cli.config.schema import DataConfig
+        from soup_cli.data.loader import load_dataset
+
+        monkeypatch.chdir(tmp_path)
+        (tmp_path / "ok.png").write_bytes(b"\x89PNG\r\n\x1a\n")
+        self._write(tmp_path / "new.jsonl", self._llava("new", "ok.png", 90))
+        self._write(
+            tmp_path / "old.jsonl",
+            self._llava("old", "../../../../../../etc/passwd", 90),
+        )
+        cfg = DataConfig(
+            train="new.jsonl", format="llava", replay="old.jsonl",
+            replay_ratio=0.1, replay_seed=0, val_split=0.0,
+        )
+        out = load_dataset(cfg)
+        escaped = [
+            row for row in out["train"]
+            if "etc/passwd" in str(row.get("image", ""))
+        ]
+        assert not escaped, (
+            "a traversal image path from the replay file reached the mix "
+            "unvalidated"
+        )
+
+    def test_replay_audio_rows_get_traversal_protection(
+        self, tmp_path, monkeypatch
+    ):
+        from soup_cli.config.schema import DataConfig
+        from soup_cli.data.loader import load_dataset
+
+        monkeypatch.chdir(tmp_path)
+        (tmp_path / "ok.wav").write_bytes(b"RIFF....WAVE")
+        self._write(
+            tmp_path / "new.jsonl",
+            [{"audio": "ok.wav", "text": f"new{i}"} for i in range(90)],
+        )
+        self._write(
+            tmp_path / "old.jsonl",
+            [
+                {"audio": "../../../../../../etc/shadow", "text": f"old{i}"}
+                for i in range(90)
+            ],
+        )
+        cfg = DataConfig(
+            train="new.jsonl", format="asr", replay="old.jsonl",
+            replay_ratio=0.1, replay_seed=0, val_split=0.0,
+        )
+        out = load_dataset(cfg)
+        escaped = [
+            row for row in out["train"]
+            if "etc/shadow" in str(row.get("audio", ""))
+        ]
+        assert not escaped, (
+            "a traversal audio path from the replay file reached the mix "
+            "unvalidated"
+        )
+
     def test_finalize_is_the_single_seam(self):
         """All three load paths must route through _finalize, or replay
         would silently apply to some datasets and not others."""
@@ -2428,6 +2565,24 @@ class TestTrainReplayFlags:
         cleaned = _clean(res.output)
         assert "--replay" in cleaned
         assert "--replay-ratio" in cleaned
+        assert "--replay-seed" in cleaned
+
+    def test_seed_override(self):
+        from soup_cli.commands.train import _apply_replay_overrides
+
+        out = _apply_replay_overrides(
+            self._cfg(_REPLAY_YAML), replay=None, replay_ratio=None,
+            replay_seed=7,
+        )
+        assert out.data.replay_seed == 7
+
+    def test_seed_without_replay_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=None, replay_seed=7
+            )
 
     def _cfg(self, yaml_str=None):
         from soup_cli.config.loader import load_config_from_string