Fix main's red Backend CI after #13850 and #13806 (#13930)

Main (bd063767f4) is red in Backend CI on five tests, seen on every PR
based on it (#13914, #13919). Four come from #13850 and #13806, which
are both right; one comes from my own #13909. Each is reproduced locally
on main and fixed here.

## From #13850 (token preview)

**`test_every_long_lived_spawner_consults_the_shutdown_latch`**:
`core/training/token_preview.py` adopts its worker with `adopt_pid` but
was in neither the gated nor the not-gated set. It already checks
`is_process_shutting_down()` before spawning, so it belongs with the
gated spawners, but a gated spawner also has to recheck after adopting:
a quit that lands between the check and `adopt_pid` happens after the
shutdown sweep took its snapshot, and the worker would outlive it.
`_spawn_locked` now rechecks right after adopting, and if the latch is
set it stops the new worker and raises `TokenPreviewError("Unsloth is
shutting down.")`, the same message `_start_locked` uses. The module
goes into the gated set.

New test `test_a_quit_that_races_the_spawn_reaps_the_new_worker` drives
that race: the latch flips inside `Process.start()`, and the worker must
be stopped and the call refused.

**`test_the_dataset_format_check_installs_the_decoder`**:
`check_format_response` now reads its rows through `load_preview_slice`,
which calls `ensure_audio_decoding()` before any tier, so the wiring is
intact. The test only read `check_format_response`'s own source. It now
checks that `check_format_response` reads through `load_preview_slice`,
and that `load_preview_slice` installs the decoder before its first read
(`resolve_dataset_path(` / `load_dataset(`), which also pins the
ordering the old form never checked.

**`test_the_subprocess_tokenizes_and_masks_typed_text`**: the
completion-masking half imports unsloth in the worker, as training does,
and Backend CI has torch but no `unsloth_zoo` (`Unsloth: Please install
unsloth_zoo via pip install unsloth_zoo then retry!`). The test skipped
only when torch was missing; it now also skips when `unsloth_zoo` is not
importable. The plain tokenization half still runs everywhere.

## From #13806 (drafter training)

**`test_owner_training_start_status_cancel_bytes_are_unchanged`**: the
training status details gained `"drafter": null`. The test pins the
exact status bytes on purpose, so a new field is a deliberate, one-line
update to the expected dict.

## From #13909 (mine)

**`test_python_children_are_told_to_emit_utf8`**: #13909 made the
drafter capability probe decode its child as UTF-8 but set
`PYTHONIOENCODING` on its own line, while this guard requires the child
env to go through `utf8_child_env()` at the call site. The probe now
passes `env = utf8_child_env(env)`, which sets the same variable through
the shared helper.

## Verification

- `test_audio_dataset_decode.py`, `test_chat_text_encoding.py`,
`test_job_accounts.py`, `test_orchestrator_unload_cancel.py`,
`test_token_preview.py`, `test_text_io_encoding.py`,
`test_drafter_training.py`: 1144 passed. One failure,
`test_capability_child_process_runs_for_real`, fails identically on main
on this host (it runs the probe in a real child interpreter) and is not
failing in CI.
- Without `unsloth_zoo` importable: the token preview real-worker test
tokenizes, then skips the masking half with its reason.
- Sabotage, the post-adopt recheck removed: the spawner-set guard fails
(`never rechecks the latch after adopting`) and the new race test fails
(`DID NOT RAISE TokenPreviewError`).
- Sabotage, `ensure_audio_decoding()` moved after
`resolve_dataset_path(` in `load_preview_slice`: the decoder test fails
(`the decoder must be installed before the first row is read`).
- `tests/test_formatter_fixed_point.py`: passes.

studio/backend/core/training/drafter.py

@@ -212,11 +212,10 @@ def _probe_capability(model_name: str, hf_token, key) -> dict:
     import sys
 
     from hub.utils.hf_tokens import apply_token_to_child_env
+    from utils.child_stdio import utf8_child_env
 
     env = dict(os.environ)
     env[_TOKEN_ENV] = "0" if hf_token is False else (hf_token or "")
-    # The parent decodes as UTF-8 below, so the child has to write it on every locale.
-    env["PYTHONIOENCODING"] = "utf-8"
     # Every Hub read in the child (config, tensor index) then runs as exactly this caller.
     apply_token_to_child_env(env, hf_token)
     result = None
@@ -224,7 +223,8 @@ def _probe_capability(model_name: str, hf_token, key) -> dict:
         proc = subprocess.run(
             [sys.executable, "-c", _PROBE.format(env = _TOKEN_ENV), model_name],
             cwd = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
-            env = env,
+            # Decoded as UTF-8 below, so the child has to write it on every locale.
+            env = utf8_child_env(env),
             capture_output = True,
             text = True,
             encoding = "utf-8",

studio/backend/core/training/token_preview.py

@@ -111,7 +111,7 @@ class _TokenPreviewService:
             native_path_secret_removed_for_child_start,
             run_without_native_path_secret,
         )
-        from utils.process_lifetime import adopt_pid
+        from utils.process_lifetime import adopt_pid, is_process_shutting_down
 
         cache_env = get_hf_cache_paths().child_env({})
         with (
@@ -139,6 +139,11 @@ class _TokenPreviewService:
             proc.start()
         adopt_pid(proc.pid)
         self._proc, self._key, self._served = proc, key, False
+        # _start_locked checked the latch before the spawn, but a quit can land in between: the
+        # shutdown sweep may already have taken its snapshot, so reap this worker here.
+        if is_process_shutting_down():
+            self._stop_locked()
+            raise TokenPreviewError("Unsloth is shutting down.")
         logger.info("Token preview worker started (pid=%s)", proc.pid)
 
     def preview(

studio/backend/tests/test_audio_dataset_decode.py

@@ -321,8 +321,18 @@ def test_the_dataset_format_check_installs_the_decoder():
 
     from hub.services.datasets import formatting
 
-    source = inspect.getsource(formatting.check_format_response)
-    assert "ensure_audio_decoding()" in source
+    # check_format_response reads its rows through load_preview_slice (#13850), so the decoder
+    # has to be installed there, ahead of every tier that can decode an audio row.
+    check = inspect.getsource(formatting.check_format_response)
+    assert "load_preview_slice(" in check, "check-format no longer reads through load_preview_slice"
+    loader = inspect.getsource(formatting.load_preview_slice)
+    body = loader[loader.index('"""', loader.index('"""') + 3) + 3 :]
+    install = body.find("ensure_audio_decoding()")
+    assert install != -1, "load_preview_slice no longer installs the audio decoder"
+    first_read = min(
+        i for i in (body.find("resolve_dataset_path("), body.find("load_dataset(")) if i != -1
+    )
+    assert install < first_read, "the decoder must be installed before the first row is read"
 
 
 def test_the_audio_trainer_paths_install_the_decoder():

studio/backend/tests/test_job_accounts.py

@@ -813,6 +813,7 @@ def test_owner_training_start_status_cancel_bytes_are_unchanged(training, monkey
             "loss": None,
             "learning_rate": None,
             "output_dir": None,
+            "drafter": None,
             "model_download_repo_id": None,
         },
         "metric_history": None,

studio/backend/tests/test_orchestrator_unload_cancel.py

@@ -3885,6 +3885,7 @@ def test_every_long_lived_spawner_consults_the_shutdown_latch():
         "core/rag/embed_llama_server.py",
         "core/systemone/native_worker.py",
         "core/systemone/laya_worker.py",
+        "core/training/token_preview.py",
         "core/training/training.py",
     }
     # Adopters this change deliberately leaves ungated, listed so the completeness check

studio/backend/tests/test_token_preview.py

@@ -177,8 +177,10 @@ def test_the_subprocess_tokenizes_and_masks_typed_text(real_worker):
     assert sample["num_tokens"] == sample["num_trained"] > 0
     assert all(isinstance(token[0], int) for token in sample["tokens"])
 
-    if importlib.util.find_spec("torch") is None:
-        pytest.skip("completion masking needs torch, as it does for training")
+    if importlib.util.find_spec("torch") is None or importlib.util.find_spec("unsloth_zoo") is None:
+        pytest.skip(
+            "completion masking imports unsloth, so it needs torch and unsloth_zoo as training does"
+        )
     chat = "<|im_start|>user\nhi<|im_end|>\n<|im_start|>assistant\nhello<|im_end|>\n"
     masked = real_worker(text = chat, train_on_completions = True, max_seq_length = 128)
     assert masked["masking"]["applied"] is True
@@ -213,6 +215,41 @@ def test_a_tokenizer_load_failure_ends_the_worker(monkeypatch):
     assert stopped[-1] is True
 
 
+def test_a_quit_that_races_the_spawn_reaps_the_new_worker(monkeypatch):
+    """The latch is checked before the spawn, but a quit can land between that check and adopt_pid,
+    after the shutdown sweep took its snapshot. The worker must not outlive it."""
+    import multiprocessing as mp_module
+
+    from core.training import account_jobs
+    from utils import process_lifetime
+
+    service = token_preview._TokenPreviewService()
+    latch = {"set": False}
+    started, stopped = [], []
+
+    class _Proc:
+        pid = 4242
+
+        def __init__(self, **kwargs):
+            pass
+
+        def start(self):
+            started.append(True)
+            latch["set"] = True  # the quit arrives while the worker is starting
+
+    monkeypatch.setattr(process_lifetime, "is_process_shutting_down", lambda: latch["set"])
+    monkeypatch.setattr(process_lifetime, "adopt_pid", lambda pid: None)
+    monkeypatch.setattr(account_jobs, "account_process_spec", lambda *args: ((), {}))
+    monkeypatch.setattr(token_preview._CTX, "Process", _Proc)
+    monkeypatch.setattr(token_preview._CTX, "Queue", lambda: mp_module.Queue())
+    monkeypatch.setattr(service, "_stop_locked", lambda: stopped.append(service._proc))
+
+    with pytest.raises(token_preview.TokenPreviewError, match = "shutting down"):
+        service._spawn_locked(("k",), {})
+    assert started == [True]
+    assert stopped and isinstance(stopped[-1], _Proc), "the adopted worker was left running"
+
+
 def test_a_supplied_cache_folder_is_never_the_tokenizer_source(captured):
     _call_route(
         TokenPreviewRequest.model_validate(