Blame for tests/test_learning.py

"""Apprentissage automatique : journalisation, format d'invite partagé, job LoRA (sur un Qwen2 minuscule, CPU)."""
import importlib.util
import json
from pathlib import Path

import pytest

from mouh.jev.types import JevMessageType as T, TaskPayload
from mouh.learning import TrajectoryRecorder, build_record, chat_messages, render_files, split_for, task_hash
from mouh.learning import prompts as P
from mouh.verification import CallableGenerator, TestTimeVerificationLoop
from mouh.verification.generators import parse_files

ROOT = Path(__file__).resolve().parents[1]
TESTS = {"test_solution.py": "from solution import add\n\ndef test_add():\n    assert add(2, 3) == 5\n"}


def load_job():
    spec = importlib.util.spec_from_file_location("train_unique", ROOT / "jobs" / "train_unique.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


JOB = load_job()


# ---------------------------------------------------------------- format partagé Space <-> job
@pytest.mark.parametrize("prev,diags", [(None, None), ({"solution.py": "x=1\n"}, ["boom", "  autre  "])])
def test_job_prompt_copy_is_identical(prev, diags):
    assert JOB.SYSTEM_PROMPT == P.SYSTEM_PROMPT
    assert JOB.chat_messages("t", TESTS, prev, diags) == P.chat_messages("t", TESTS, prev, diags)
    files = {"solution.py": "def add(a, b):\n    return a + b\n", "pkg/util.py": "X = 1\n"}
    assert JOB.render_files(files) == P.render_files(files)


def test_rendered_target_parses_back():
    files = {"solution.py": "def add(a, b):\n    return a + b\n", "pkg/util.py": "X = 1\n"}
    assert parse_files(render_files(files)) == files
    assert JOB.parse_files(render_files(files)) == files


# ---------------------------------------------------------------- bus et enregistrement
def run_with_repair():
    calls = []

    def gen(req):
        calls.append(req)
        return {"solution.py": "def add(a, b):\n    return a + b\n"}

    loop = TestTimeVerificationLoop(CallableGenerator(gen), max_attempts=3)
    task = TaskPayload(instruction="Écrire add", candidate_files={"solution.py": "def add(a, b):\n    return a - b\n"},
                       test_files=TESTS)
    return task, loop.run(task), calls


def test_router_decision_precedes_code_proposal():
    _, out, calls = run_with_repair()
    types = [m.type for m in out.messages]
    for i, t in enumerate(types):
        if t == T.CODE_PROPOSAL:
            assert types[i - 1] == T.ROUTE_DECISION, types
    assert calls and calls[0].experts  # le générateur reçoit les experts décidés par le routeur


def test_build_record_keeps_attempts_and_repair():
    task, out, _ = run_with_repair()
    rec = build_record(task, out, "test")
    assert rec["certified"] and rec["status"] == "certified"
    assert [a["certified"] for a in rec["attempts"]] == [False, True]
    assert rec["attempts"][0]["diagnostics"] and "a - b" in rec["attempts"][0]["files"]["solution.py"]
    assert rec["final_files"]["solution.py"].endswith("a + b\n")
    assert rec["split"] == split_for(rec["task_hash"]) and rec["task_hash"] == task_hash(task.instruction, TESTS)
    json.dumps(rec)  # sérialisable


def test_task_hash_ignores_whitespace_and_split_is_deterministic():
    assert task_hash("a  b", {"t.py": "x =  1"}) == task_hash("a b", {"t.py": "x = 1\n"})
    splits = {split_for(task_hash(f"t{i}", TESTS)) for i in range(200)}
    assert splits == {"train", "heldout"}


class FakeScheduler:
    def __init__(self, **kw):
        import threading

        self.kw, self.lock = kw, threading.Lock()


def test_recorder_local_and_remote(tmp_path):
    local = TrajectoryRecorder(tmp_path / "a")
    assert not local.remote
    local.record({"x": 1})
    assert json.loads(local.path.read_text()) == {"x": 1}
    remote = TrajectoryRecorder(tmp_path / "b", "me/ds", "tok", every_min=3, scheduler_factory=FakeScheduler)
    assert remote.remote and remote._scheduler.kw["private"] and remote._scheduler.kw["repo_type"] == "dataset"
    assert remote._scheduler.kw["allow_patterns"] == ["data/*.jsonl"]


def test_service_records_session(tmp_path, monkeypatch):
    import mouh.learning.recorder as R
    from mouh.service import stream_session

    monkeypatch.setattr(R, "_RECORDER", TrajectoryRecorder(tmp_path))
    snaps = list(stream_session("add", "def add(a, b):\n    return a + b\n", TESTS["test_solution.py"]))
    rep = json.loads(snaps[-1]["report"])
    assert rep["journalisation"].startswith("oui")
    rec = json.loads(R._RECORDER.path.read_text().splitlines()[-1])
    assert rec["certified"] and rec["schema"] == "mouh-session/1"
    snaps = list(stream_session("add", "def add(a, b):\n    return a + b\n", TESTS["test_solution.py"], record=False))
    assert json.loads(snaps[-1]["report"])["journalisation"].startswith("non")
    assert len(R._RECORDER.path.read_text().splitlines()) == 1


# ---------------------------------------------------------------- données du job
def session(i, *, certified=True, tests=None, split=None, failed_first=True):
    tests = tests or {"test_solution.py": f"from solution import f\n\ndef test_f():\n    assert f() == {i}\n"}
    th = task_hash(f"tâche {i}", tests)
    final = {"solution.py": f"def f():\n    return {i}\n"}
    attempts = []
    if failed_first:
        attempts.append({"attempt": 1, "origin": "user", "files": {"solution.py": "def f():\n    return -1\n"},
                         "score": 0.7, "certified": False, "diagnostics": [f"assert -1 == {i}"]})
    attempts.append({"attempt": len(attempts) + 1, "origin": "model", "files": final, "score": 1.0,
                     "certified": certified, "diagnostics": []})
    return {"schema": "mouh-session/1", "session_id": f"s{i}", "task_hash": th, "split": split or split_for(th),
            "instruction": f"tâche {i}", "tests": tests, "proofs": [], "generator": "x",
            "status": "certified" if certified else "exhausted", "certified": certified,
            "final_files": final if certified else None, "attempts": attempts}


def test_build_examples_filters_and_separates_heldout():
    sessions = [session(1, split="train"), session(2, split="heldout"), session(3, certified=False, split="train"),
                session(4, split="train", tests={"test_solution.py": "def test_x():\n    assert True\n"}),
                session(1, split="train")]  # doublon exact
    train, heldout, stats = JOB.build_examples(sessions)
    assert stats["kept"] == 3 and [t["task_hash"] for t in heldout] == [sessions[1]["task_hash"]]
    assert sorted(e["kind"] for e in train) == ["fix", "gen"]  # doublon retiré, tâche 4 rejetée (n'importe pas solution)
    assert all(sessions[1]["instruction"] not in json.dumps(e) for e in train)


def test_run_tests_really_executes():
    t = {"test_solution.py": "from solution import f\n\ndef test_f():\n    assert f() == 3\n"}
    assert JOB.run_tests({"solution.py": "def f():\n    return 3\n"}, t)
    assert not JOB.run_tests({"solution.py": "def f():\n    return 4\n"}, t)
    assert not JOB.run_tests({"solution.py": "import time\ndef f():\n    time.sleep(5)\n"}, t, timeout_s=2)


# ---------------------------------------------------------------- job complet sur un modèle minuscule
@pytest.fixture(scope="module")
def tiny_qwen(tmp_path_factory):
    pytest.importorskip("peft")
    from tokenizers import Tokenizer, decoders, models, pre_tokenizers, trainers
    from transformers import PreTrainedTokenizerFast, Qwen2Config, Qwen2ForCausalLM

    d = tmp_path_factory.mktemp("tiny_qwen")
    tk = Tokenizer(models.BPE())
    tk.pre_tokenizer = pre_tokenizers.ByteLevel(add_prefix_space=False)
    tk.decoder = decoders.ByteLevel()
    corpus = [P.SYSTEM_PROMPT, render_files({"solution.py": "def f():\n    return 1\n"}), "## Tâche ## Tests assert"] * 20
    tk.train_from_iterator(corpus, trainers.BpeTrainer(vocab_size=400, special_tokens=["<|endoftext|>", "<|im_start|>", "<|im_end|>"],
                                                        initial_alphabet=pre_tokenizers.ByteLevel.alphabet()))
    tok = PreTrainedTokenizerFast(tokenizer_object=tk, eos_token="<|im_end|>", pad_token="<|endoftext|>")
    tok.chat_template = ("{% for m in messages %}<|im_start|>{{ m['role'] }}\n{{ m['content'] }}<|im_end|>\n{% endfor %}"
                         "{% if add_generation_prompt %}<|im_start|>assistant\n{% endif %}")
    tok.save_pretrained(d)
    cfg = Qwen2Config(vocab_size=len(tok), hidden_size=32, intermediate_size=64, num_hidden_layers=2,
                      num_attention_heads=4, num_key_value_heads=2, max_position_embeddings=4096,
                      eos_token_id=tok.eos_token_id, pad_token_id=tok.pad_token_id, tie_word_embeddings=True)
    Qwen2ForCausalLM(cfg).save_pretrained(d)
    return str(d)


def write_sessions(path: Path, n: int, start: int = 0):
    path.mkdir(parents=True, exist_ok=True)
    with open(path / f"s{start}.jsonl", "w") as fh:
        for i in range(start, start + n):
            fh.write(json.dumps(session(i)) + "\n")


def test_unique_generator_runs_in_the_loop(tiny_qwen, monkeypatch):
    import mouh.service as S

    monkeypatch.setenv("UNIQUE_BASE_MODEL", tiny_qwen)
    monkeypatch.setenv("UNIQUE_MAX_NEW_TOKENS", "8")
    monkeypatch.delenv("UNIQUE_ADAPTER_REPO", raising=False)
    S._MODEL_CACHE.pop("unique", None)
    snaps = list(S.stream_session("add", "def add(a, b):\n    return a - b\n", TESTS["test_solution.py"], max_attempts=2,
                                  generator_choice=S.GENERATOR_CHOICES[2], record=False))
    assert snaps[-1]["status"].startswith("AUCUNE SORTIE")  # modèle aléatoire : rien de non prouvé n'est émis
    S._MODEL_CACHE.pop("unique", None)


# ---------------------------------------------------------------- pilote automatique (tout sur HF)


def test_app_builds():
    import importlib

    import app  # noqa: F401 — l'interface se construit sans réseau ni jeton

    importlib.reload(app)


# ---------------------------------------------------------------- alimentation en masse


# ---------------------------------------------------------------- boucle continue (TRL SFT + GRPO)
FAST = ["--min-heldout", "1", "--sft-chunk", "8", "--batch-size", "2", "--grad-accum", "1", "--max-len", "1024",
        "--grpo-steps", "2", "--grpo-batch", "4", "--grpo-generations", "2", "--grpo-max-completion", "16",
        "--max-new-tokens", "8", "--progress-every", "0", "--benchmark", "0", "--memory-ab", "0",
        "--promotion-margin", "1", "--measure-best-of", "0"]


def test_pytest_reward_scores_real_execution():
    r = JOB.make_pytest_reward(workers=2)
    tests = json.dumps({"test_solution.py": "from solution import f\n\ndef test_f():\n    assert f() == 3\n"})
    good = [{"role": "assistant", "content": render_files({"solution.py": "def f():\n    return 3\n"})}]
    bad = render_files({"solution.py": "def f():\n    return 4\n"})
    assert r(completions=[good, bad], tests=[tests] * 2) == [1.0, 0.1]
    assert r(completions=["### FILE: solution.py\n```python\ndef (\n```\n### END"], tests=[tests]) == [0.0]


def test_continuous_loop_runs_sft_and_grpo_with_trl(tiny_qwen, tmp_path):
    pytest.importorskip("trl")
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 60)
    assert JOB.main(["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1",
                     *FAST]) == 0
    st = json.loads((out / "state.json").read_text())
    assert [h["phase"] for h in st["history"]] == ["sft", "grpo"]
    assert st["history"][0]["train"]["steps"] > 0 and st["history"][1]["train"]["steps"] == 2
    assert not any(h["promoted"] for h in st["history"])  # modèle aléatoire : rien n'est promu
    prog = json.loads((out / "progress.json").read_text())
    assert prog["phase"] == "fin" and prog["cycle"] == 1 and len(prog["history"]) == 2
    assert prog["base"]["n"] == len(st["heldout"]) and "rewards/pytest_reward/mean" in prog["metrics"]


def test_gate_promotes_only_strict_improvement_and_heldout_is_frozen(tiny_qwen, tmp_path, monkeypatch):
    pytest.importorskip("trl")
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 60)
    scores = iter([1, 3, 2,   4, 4])  # passage 1 : base 1, SFT 3 (promu), GRPO 2 ; passage 2 : base en cache, SFT 4 (promu), GRPO 4
    monkeypatch.setattr(JOB, "pass_at_1", lambda *a, **k: {"passed": next(scores), "n": 5, "rate": 0, "ci95": [0, 1]})
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1",
            "--max-heldout", "3", *FAST]
    assert JOB.main(args) == 0
    st1 = json.loads((out / "state.json").read_text())
    assert [h["promoted"] for h in st1["history"]] == [True, False] and (out / "adapter" / "adapter_config.json").exists()
    write_sessions(data, 20, start=500)  # nouvelles données : le heldout figé ne change pas
    assert JOB.main(args) == 0
    st2 = json.loads((out / "state.json").read_text())
    assert st2["heldout"] == st1["heldout"] and st2["base"]["passed"] == 1
    assert [h["promoted"] for h in st2["history"]] == [True, False, True, False] and st2["best"]["passed"] == 4
    assert st2["sft_cursor"] == 16  # la 2e phase SFT a pris la tranche suivante


def test_daily_gpu_budget_pauses_the_loop(tiny_qwen, tmp_path):
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 30)
    assert JOB.main(["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen,
                     "--daily-gpu-hours", "0", *FAST]) == 0
    prog = json.loads((out / "progress.json").read_text())
    assert prog["phase"] == "pause" and "budget" in prog["status"] and not (out / "state.json").exists()


def test_waits_when_not_enough_heldout(tiny_qwen, tmp_path):
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 30)
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, *FAST]
    args[args.index("--min-heldout") + 1] = "1000"
    assert JOB.main(args) == 0
    assert json.loads((out / "progress.json").read_text())["phase"] == "attente"


def test_heldout_extension_remeasures_current_adapter(tiny_qwen, tmp_path, monkeypatch):
    pytest.importorskip("trl")
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 40)
    scores = iter([1, 3, 0,   2, 5, 4, 4])  # passage 2 (heldout agrandi) : base 2, actuel 5, SFT 4 et GRPO 4 -> rien promu
    monkeypatch.setattr(JOB, "pass_at_1", lambda *a, **k: {"passed": next(scores), "n": 9, "rate": 0, "ci95": [0, 1]})
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1", *FAST]
    assert JOB.main(args) == 0
    write_sessions(data, 60, start=300)
    assert JOB.main(args) == 0
    st = json.loads((out / "state.json").read_text())
    assert st["best"]["passed"] == 5 and st["best"]["source"] == "actuel"
    assert [h["promoted"] for h in st["history"]] == [True, False, False, False]


# ---------------------------------------------------------------- mémoire à long terme
from mouh.learning.memory import TaskMemory, items_from_records  # noqa: E402


def mem_items():
    base = [("Écrire une fonction qui inverse une chaîne de caractères", "def f(s):\n    return s[::-1]\n"),
            ("Compter les voyelles dans une chaîne", "def f(s):\n    return sum(c in 'aeiou' for c in s)\n"),
            ("Calculer la somme des éléments pairs d'une liste", "def f(l):\n    return sum(x for x in l if x % 2 == 0)\n")]
    return [{"task_hash": f"h{i}", "instruction": t, "files": {"solution.py": c}} for i, (t, c) in enumerate(base)]


def test_memory_finds_similar_but_never_the_task_itself_or_near_duplicates():
    m = TaskMemory(mem_items())
    hits = m.search("Compter le nombre de voyelles d'un texte", k=1)
    assert hits and hits[0]["task_hash"] == "h1"
    assert all(h["task_hash"] != "h1" for h in m.search("Compter les voyelles dans une chaîne", k=3, exclude_hash="h1"))
    assert all(h["task_hash"] != "h0" for h in m.search("Écrire une fonction qui inverse une chaîne de caractères", k=3))
    jm = JOB.TaskMemory(mem_items())
    for q in ("voyelles texte", "somme pairs liste", "inverse chaîne"):
        assert [h["task_hash"] for h in jm.search(q, 2)] == [h["task_hash"] for h in m.search(q, 2)]


def test_memory_only_uses_certified_train_records():
    recs = [session(i) for i in range(30)] + [session(99, certified=False)]
    items = items_from_records(recs)
    assert items and all(r["split"] == "train" for r in recs if r["task_hash"] in {i["task_hash"] for i in items})
    assert session(99)["task_hash"] not in {i["task_hash"] for i in items}


def test_prompt_with_examples_identical_in_job():
    ex = mem_items()[:2]
    assert JOB.chat_messages("t", TESTS, examples=ex) == P.chat_messages("t", TESTS, examples=ex)
    assert "Exemples validés similaires" in P.user_prompt("t", TESTS, examples=ex)
    assert P.user_prompt("t", TESTS) == P.user_prompt("t", TESTS, examples=[])


# ---------------------------------------------------------------- curriculum + vrais échecs
def test_curriculum_prefers_informative_tasks():
    import random as R

    tasks = [{"task_hash": f"t{i}"} for i in range(300)]
    stats = {f"t{i}": ([4, 4] if i < 100 else [0, 10] if i < 200 else [2, 4]) for i in range(300)}
    picked = JOB.curriculum_select(tasks, stats, 60, R.Random(0))
    zone = sum(1 for t in picked if int(t["task_hash"][1:]) >= 200)
    assert len(picked) == 60 and zone >= 45  # surtout des tâches réussies « parfois »


def test_reward_logs_outcomes_and_real_failures():
    log_ = JOB.RolloutLog()
    r = JOB.make_pytest_reward(workers=2, log_to=log_)
    tests = json.dumps({"test_solution.py": "from solution import f\n\ndef test_f():\n    assert f() == 3\n"})
    good = render_files({"solution.py": "def f():\n    return 3\n"})
    bad = render_files({"solution.py": "def f():\n    return 4\n"})
    assert r(completions=[good, bad], tests=[tests] * 2, task_hash=["h", "h"], instruction=["énoncé", "énoncé"]) == [1.0, 0.1]
    assert log_.outcomes["h"] == [1, 2]
    (f,) = log_.failures
    assert "assert 4 == 3" in f["pytest_output"] and f["files"]["solution.py"].strip().endswith("return 4")


def test_decontamination_drops_benchmark_lookalikes():
    bench = [{"instruction": "Complète : def has_close_elements(numbers, threshold): Check if in given list of numbers, "
                             "are any two numbers closer to each other than given threshold."}]
    s1 = session(1)
    s2 = {**session(2), "instruction": "Check if in given list of numbers, are any two numbers closer to each other "
                                       "than given threshold. Return bool."}
    kept, removed = JOB.decontaminate([s1, s2], bench)
    assert removed == 1 and kept == [s1]


def humaneval_rows():
    return [{"task_id": f"HE/{i}", "entry_point": f"plus{i}",
             "prompt": f"def plus{i}(x):\n    \"\"\"Ajoute {i}.\"\"\"\n",
             "test": f"def check(candidate):\n    assert candidate(1) == {1 + i}\n"} for i in range(3)]


def test_humaneval_conversion_runs_real_checks(tmp_path):
    p = tmp_path / "he.jsonl"
    p.write_text("\n".join(json.dumps(r) for r in humaneval_rows()))
    t = JOB.load_humaneval(str(p), None)[1]
    assert JOB.run_tests({"solution.py": "def plus1(x):\n    return x + 1\n"}, t["tests"])
    assert not JOB.run_tests({"solution.py": "def plus1(x):\n    return x\n"}, t["tests"])


def test_loop_with_benchmark_memory_failures_and_repairs(tiny_qwen, tmp_path, monkeypatch):
    pytest.importorskip("trl")
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 60)
    he = tmp_path / "he.jsonl"
    he.write_text("\n".join(json.dumps(r) for r in humaneval_rows()))
    # le modèle minuscule ne réussit rien : on simule des échecs réels pour vérifier le circuit complet
    real = JOB.make_pytest_reward

    def fake_reward(workers=8, timeout_s=20.0, log_to=None):
        def rw(prompts=None, completions=None, tests=None, task_hash=None, instruction=None, **kw):
            for th, ins, t in zip(task_hash, instruction, tests):
                if ins:
                    log_to.add(th, False, {"task_hash": th, "instruction": ins, "tests": json.loads(t),
                                           "files": {"solution.py": "def f():\n    return -1\n"},
                                           "pytest_output": "E   assert -1 == 1"})
            return [0.0] * len(completions)
        return rw

    monkeypatch.setattr(JOB, "make_pytest_reward", fake_reward)
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1", *FAST]
    args[args.index("--benchmark") + 1] = "1"
    args[args.index("--memory-ab") + 1] = "1"
    args += ["--local-benchmark", str(he)]
    assert JOB.main(args) == 0
    st = json.loads((out / "state.json").read_text())
    m = st["measures"]
    assert m["humaneval"]["n"] == 3 and m["humaneval_base"]["n"] == 3 and m["memory"]["with"]["n"] == len(st["heldout"])
    assert st["history"][1]["train"]["curriculum"]["failures"] > 0 and st["task_stats"]
    assert list((out / "failures").glob("*.jsonl"))
    assert JOB.main(args) == 0  # 2e passage : les échecs réels deviennent des prompts de réparation GRPO
    st2 = json.loads((out / "state.json").read_text())
    assert st2["history"][-1]["train"]["repairs"] > 0
    assert json.loads((out / "progress.json").read_text())["measures"]["humaneval"]["n"] == 3
    monkeypatch.setattr(JOB, "make_pytest_reward", real)


# ---------------------------------------------------------------- modèle maître


def test_sft_order_puts_the_teachers_very_hard_tasks_first_and_keeps_the_old_order_without_levels():
    import hashlib

    def rec(i, generator, level):
        attempt = {"attempt": 1, "origin": "dataset", "files": {"solution.py": "x"}, "score": 1.0,
                   "certified": True, "diagnostics": []}
        if level:
            attempt["level"] = level
        return {"schema": "mouh-session/1", "certified": True, "split": "train", "task_hash": f"h{i}",
                "instruction": f"tâche {i}", "generator": generator,
                "tests": {"test_solution.py": "from solution import f\n\n\ndef test_a():\n    assert f(1) == 1\n"},
                "final_files": {"solution.py": f"def f(x):\n    return {i}\n"}, "attempts": [attempt]}

    recs = [rec(i, "ingest:kodcode", None) for i in range(30)] + [
        rec(100, "teacher:m", "extrême"), rec(101, "teacher:m", "difficile"), rec(102, "teacher:m", "facile")]
    train, _, stats = JOB.build_examples(recs)
    order = JOB.sft_order(train, 0)
    tiers = [train[i]["tier"] for i in order]
    assert tiers[0] == 0 and tiers[1] == 1 and tiers.count(0) == 1 and stats["top_teacher"] == 1
    assert [train[i]["tier"] for i in order[2:]] == [2] * (len(train) - 2)
    # sans niveaux, l'ordre est exactement celui d'avant v3.11
    plain = [{k: v for k, v in e.items() if k != "tier"} for e in train[:30]]
    old = sorted(range(len(plain)), key=lambda i: hashlib.sha256(
        f"0:{JOB._sha(plain[i]['target'])}:{plain[i]['kind']}:{i}".encode()).hexdigest())
    assert JOB.sft_order(plain, 0) == old


# ---------------------------------------------------------------- interrupteurs (retour arrière)


def test_phase_error_is_reported_and_loop_continues(tiny_qwen, tmp_path, monkeypatch):
    pytest.importorskip("trl")
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 60)

    def boom(*a, **k):
        raise ValueError("bf16 non pris en charge")

    monkeypatch.setattr(JOB, "run_sft", boom)
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1", *FAST]
    assert JOB.main(args) == 0
    st = json.loads((out / "state.json").read_text())
    assert st["last_error"]["phase"] == "sft" and "bf16" in st["last_error"]["error"]
    assert [h["phase"] for h in st["history"]] == ["grpo"]  # GRPO a quand même tourné
    monkeypatch.setattr(JOB, "run_grpo", boom)
    assert JOB.main(args) == 0  # 2 échecs d'affilée (SFT puis GRPO) : on continue
    assert JOB.main(args) == 1  # 3e échec d'affilée : arrêt plutôt que de payer le GPU pour rien
    assert json.loads((out / "progress.json").read_text())["phase"] == "erreur"


def test_precision_never_requests_bf16_on_t4(monkeypatch):
    import torch

    monkeypatch.setattr(torch.cuda, "is_available", lambda: True)
    monkeypatch.setattr(torch.cuda, "is_bf16_supported", lambda *a, **k: True)  # ce que PyTorch répond sur T4
    monkeypatch.setattr(torch.cuda, "get_device_capability", lambda *a: (7, 5))
    assert JOB._precision() == {"fp16": True, "bf16": False}
    monkeypatch.setattr(torch.cuda, "get_device_capability", lambda *a: (8, 9))
    assert JOB._precision() == {"bf16": True, "fp16": False}


def test_gpu_compat_disables_16_to_32_gemm_on_t4(monkeypatch):
    import torch
    from trl.kernels import chunked_logprob

    monkeypatch.setattr(torch.cuda, "is_available", lambda: True)
    monkeypatch.setattr(torch.cuda, "get_device_capability", lambda *a: (7, 5))
    monkeypatch.setattr(chunked_logprob, "_MM_OUT_DTYPE", True)
    JOB._gpu_compat()
    assert chunked_logprob._MM_OUT_DTYPE is False
    acc = torch.zeros(2, 2)
    chunked_logprob._addmm_fp32(acc, torch.ones(2, 3, dtype=torch.float16), torch.ones(3, 2, dtype=torch.float16))
    assert torch.equal(acc, torch.full((2, 2), 3.0))  # chemin de repli fp32


# ---------------------------------------------------------------- v3.5
def test_no_signal_stopper_stops_after_patience():
    import types

    info = {}
    cb = JOB._no_signal_stopper(0.9, 3, 2, info)
    state, control = types.SimpleNamespace(global_step=0), types.SimpleNamespace(should_training_stop=False)
    for step, z in enumerate([1.0, 0.5, 1.0, 1.0, 0.95], 1):
        state.global_step = step
        cb.on_log(None, state, control, logs={"frac_reward_zero_std": z})
    assert control.should_training_stop and info["early_stop"]["step"] == 5


def test_curriculum_prefers_hard_tasks_and_drops_solved_ones():
    import random as R

    tasks = [{"task_hash": f"t{i}", "level": "hard" if i < 50 else "easy"} for i in range(100)]
    picked = JOB.curriculum_select(tasks, {}, 30, R.Random(1))
    assert sum(t["level"] == "hard" for t in picked) >= 24
    solved = {f"t{i}": [2, 2] for i in range(50)}  # les difficiles déjà réussies 2 fois sur 2
    picked = JOB.curriculum_select(tasks, solved, 30, R.Random(1))
    assert sum(t["level"] == "hard" for t in picked) <= 12


def test_grpo_stops_early_when_every_attempt_has_the_same_score(tiny_qwen, tmp_path, monkeypatch):
    pytest.importorskip("trl")
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 60)

    def always_one(workers=8, timeout_s=20.0, log_to=None):
        return lambda completions=None, **kw: [1.0] * len(completions)

    monkeypatch.setattr(JOB, "make_pytest_reward", always_one)
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1", *FAST,
            "--grpo-stop-patience", "2", "--grpo-stop-min-steps", "2"]
    args[args.index("--grpo-steps") + 1] = "8"
    assert JOB.main(args) == 0
    grpo = json.loads((out / "state.json").read_text())["history"][-1]["train"]
    assert grpo["steps"] < 8 and grpo["early_stop"]["step"] == grpo["steps"]


# ---------------------------------------------------------------- v3.6 : snapshot partagé data/ + failures/
def shared_snapshot(tmp_path):
    root = tmp_path / "snapshot"
    (root / "data").mkdir(parents=True)
    (root / "failures").mkdir()
    (root / "state").mkdir()
    (root / "data" / "ingest-x.jsonl").write_text("\n".join(json.dumps(session(i)) for i in range(5)) + "\n")
    fail = {"task_hash": "h", "instruction": "i", "tests": TESTS, "files": {"solution.py": "x=1\n"},
            "pytest_output": "E assert"}
    (root / "failures" / "r-c1.jsonl").write_text(json.dumps(fail) + "\n" + json.dumps({"incomplet": 1}) + "\n")
    (root / "state" / "teacher.json").write_text("{}")
    return root


def test_loaders_read_only_their_own_folder_of_a_shared_snapshot(tmp_path, monkeypatch):
    import huggingface_hub

    root = shared_snapshot(tmp_path)
    monkeypatch.setattr(huggingface_hub, "snapshot_download", lambda *a, **k: str(root))

    class ListApi:
        def __init__(self, token=None):
            pass

        def list_repo_files(self, repo, repo_type=None):
            return [str(f.relative_to(root)) for f in root.rglob("*") if f.is_file()]

    monkeypatch.setattr(huggingface_hub, "HfApi", ListApi)
    monkeypatch.setattr(huggingface_hub, "hf_hub_download", lambda repo, name, **k: str(root / name))
    sessions = JOB.load_sessions(None, "me/ds", "tok")
    assert len(sessions) == 5 and all(r["schema"] == "mouh-session/1" for r in sessions)
    store = types_store = type("S", (), {"local": None})()
    fails = JOB.load_failures(store, "me/ds", "tok", 100)
    assert len(fails) == 1 and fails[0]["files"]  # ni les sessions de data/, ni l'enregistrement incomplet


# ---------------------------------------------------------------- v3.7
def test_select_data_files_keeps_recent_ingest_slices_only():
    names = [f"data/ingest-opencodeinstruct-{i * 4000:09d}.jsonl" for i in range(50)] + \
            [f"data/ingest-kodcode-{i * 2000:09d}.jsonl" for i in range(3)] + \
            ["data/teacher-20261104T090000Z.jsonl", "data/sessions-abc.jsonl"]
    chosen = JOB.select_data_files(names, 30)
    oci = [n for n in chosen if "opencodeinstruct" in n]
    assert len(oci) == 30 and oci[0].endswith(f"{20 * 4000:09d}.jsonl")  # les 30 plus récentes
    assert sum("kodcode" in n for n in chosen) == 3
    assert "data/teacher-20261104T090000Z.jsonl" in chosen and "data/sessions-abc.jsonl" in chosen
    assert len(JOB.select_data_files(names, 0)) == len(names)


def test_frozen_heldout_survives_when_its_files_are_no_longer_loaded(tiny_qwen, tmp_path, monkeypatch):
    pytest.importorskip("trl")
    data, out = tmp_path / "data", tmp_path / "out"
    data.mkdir()

    def slice_(start, n, off):
        (data / f"ingest-opencodeinstruct-{off:09d}.jsonl").write_text(
            "\n".join(json.dumps(session(i)) for i in range(start, start + n)) + "\n")

    slice_(0, 60, 0)
    scores = iter([5, 6, 6, 7, 7])  # base 5 ; puis seulement des candidats : jamais de re-mesure de la base
    monkeypatch.setattr(JOB, "pass_at_1", lambda *a, **k: {"passed": next(scores), "n": 6, "rate": 0, "ci95": [0, 1]})
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1", *FAST,
            "--max-ingest-files", "1", "--max-heldout", "3"]
    assert JOB.main(args) == 0
    st1 = json.loads((out / "state.json").read_text())
    slice_(500, 60, 4000)  # nouvelle tranche : avec --max-ingest-files 1, la première n'est plus chargée
    assert JOB.main(args) == 0
    st2 = json.loads((out / "state.json").read_text())
    assert st2["heldout"] == st1["heldout"] and st2["heldout_sig"] == st1["heldout_sig"]
    assert set(st2["heldout_tasks"]) == set(st1["heldout"]) and st2["base"]["passed"] == 5


def test_promotion_needs_the_margin(tiny_qwen, tmp_path, monkeypatch):
    pytest.importorskip("trl")
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 60)
    scores = iter([60, 62, 63])  # base 60 ; SFT 62 (+2) écarté ; GRPO 63 (+3) mis en service
    monkeypatch.setattr(JOB, "pass_at_1", lambda *a, **k: {"passed": next(scores), "n": 100, "rate": 0, "ci95": [0, 1]})
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1", *FAST]
    args[args.index("--promotion-margin") + 1] = "3"
    assert JOB.main(args) == 0
    assert [h["promoted"] for h in json.loads((out / "state.json").read_text())["history"]] == [False, True]


# ---------------------------------------------------------------- v3.8


def test_shrink_batches_keeps_effective_batch():
    import types

    a = types.SimpleNamespace(batch_size=2, grad_accum=8, grpo_batch=8, grpo_generations=4, grpo_max_completion=384,
                              eval_batch=16)
    t = JOB.shrink_batches(a, "sft")
    assert (t["batch_size"], t["grad_accum"]) == (1, 16)
    t = JOB.shrink_batches(a, "grpo")
    assert t["grpo_batch"] == 4 and t["eval_batch"] == 4
    t = JOB.shrink_batches(a, "grpo")
    assert t["grpo_batch"] == 4 and t["grpo_max_completion"] == 192


def test_out_of_memory_shrinks_batches_for_next_runs(tiny_qwen, tmp_path, monkeypatch):
    pytest.importorskip("trl")
    import torch

    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 60)
    real = JOB.run_sft

    def oom(*a, **k):
        raise torch.cuda.OutOfMemoryError("CUDA out of memory. Tried to allocate 2.00 GiB")

    monkeypatch.setattr(JOB, "run_sft", oom)
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1", *FAST]
    assert JOB.main(args) == 0
    st = json.loads((out / "state.json").read_text())
    assert st["tuning"]["batch_size"] == 1 and st["tuning"]["grad_accum"] == 2
    seen = {}
    monkeypatch.setattr(JOB, "run_sft", lambda a, *r, **k: seen.update(bs=a.batch_size) or real(a, *r, **k))
    assert JOB.main(args) == 0 and seen["bs"] == 1  # la réduction est reprise au passage suivant


def test_phase_that_cannot_fit_is_left_for_next_run(tiny_qwen, tmp_path):
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 40)
    out.mkdir()
    (out / "state.json").write_text(json.dumps({"phase_seconds": {"sft": 10_000_000}}))
    assert JOB.main(["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, *FAST]) == 0
    assert json.loads((out / "state.json").read_text()).get("history", []) == []


def test_progress_history_is_per_model(tmp_path):
    store = JOB.Store(local_out=str(tmp_path), adapter_repo="me/a", bucket="me/b", token=None)
    store.write_progress({"adapter": "me/ancien", "day": JOB._now().strftime("%Y-%m-%d"), "gpu_hours_today": 3.0,
                          "history": [{"x": 1}], "best": {"passed": 9}}, True)
    p = JOB.Progress(store, run_id="r", max_hours=1, daily_budget_h=24, min_interval_s=0, adapter="me/nouveau")
    assert p.state["history"] == [] and p.state["best"] is None and p.state["gpu_hours_before_run"] == 3.0


def test_best_of_measure_runs_and_is_reported(tiny_qwen, tmp_path):
    pytest.importorskip("trl")
    data, out = tmp_path / "data", tmp_path / "out"
    write_sessions(data, 40)
    args = ["--local-data", str(data), "--local-out", str(out), "--base-model", tiny_qwen, "--cycles", "1", *FAST,
            "--measure-best-of", "2"]
    assert JOB.main(args) == 0
    m = json.loads((out / "state.json").read_text())["measures"]
    assert m["best_of"]["k"] == 2 and m["best_of"]["n"] > 0


def test_prompt_notes_identical_in_job():
    notes = ["doc : utiliser heapq", "autre"]
    assert JOB.chat_messages("t", TESTS, notes=notes) == P.chat_messages("t", TESTS, notes=notes)
    assert "Notes de recherche web" in P.user_prompt("t", TESTS, notes=notes)


def test_complexity_estimate_orders_tasks():
    from mouh.learning.unique_generator import estimate_complexity

    easy = estimate_complexity("Écrire add(a, b).", TESTS)
    hard_tests = {"test_solution.py": "from solution import shortest, Graph\n" +
                  "".join(f"def test_{i}():\n    assert shortest(Graph(), {i}) == {i}\n" for i in range(14))}
    hard = estimate_complexity("Implémenter Dijkstra sur un graphe orienté pondéré, complexité O(n log n), "
                               "avec un tas (heap) et une classe Graph. " * 6, hard_tests)
    assert easy["level"] == "simple" and hard["level"] == "complexe" and hard["score"] > easy["score"]


def test_adaptive_session_picks_the_candidate_that_passes(monkeypatch):
    from mouh.learning import unique_generator as U
    from mouh.verification.generators import GenerationRequest
    from mouh.jev.types import JevMessageType as MT

    class Gen:
        def build_messages(self, r):
            return [{"role": "user", "content": "x"}]

        def candidates(self, messages, n):
            Gen.n = n
            return [{"solution.py": "def add(a, b):\n    return a - b\n"},
                    {"solution.py": "def add(a, b):\n    return a + b\n"}][:max(n, 1)]

    monkeypatch.setattr(U, "estimate_complexity", lambda i, t: {"level": "complexe", "score": 0.9})
    task = TaskPayload(instruction="add", test_files=TESTS)
    sess = U.UniqueSession(Gen(), adaptive=True)
    out = sess.generate(GenerationRequest(task=task, attempt=1, message_type=MT.TASK_SUBMISSION, experts=(1,)))
    assert out["solution.py"].endswith("a + b\n") and Gen.n == 6
    assert sess.info["essais"][0]["retenue"] == 2 and "recherche_web" not in sess.info   # v4.1 : aucune recherche web
    assert not hasattr(U, "web_notes")
    calm = U.UniqueSession(Gen(), adaptive=False)
    calm.generate(GenerationRequest(task=task, attempt=1, message_type=MT.TASK_SUBMISSION, experts=(1,)))
    assert Gen.n == 1


def test_reflective_generator_end_to_end(tiny_qwen, monkeypatch):
    import mouh.service as S

    monkeypatch.setenv("UNIQUE_BASE_MODEL", tiny_qwen)
    monkeypatch.setenv("UNIQUE_MAX_NEW_TOKENS", "8")
    monkeypatch.setenv("UNIQUE_MAX_SAMPLES", "2")
    monkeypatch.delenv("UNIQUE_ADAPTER_REPO", raising=False)
    S._MODEL_CACHE.pop("unique", None)
    snaps = list(S.stream_session("add", "def add(a, b):\n    return a - b\n", TESTS["test_solution.py"], max_attempts=2,
                                  generator_choice=S.GENERATOR_CHOICES[3], record=False))
    rep = json.loads(snaps[-1]["report"])
    assert snaps[-1]["status"].startswith("AUCUNE SORTIE") and rep["reflexion"]["complexite"]["level"] == "simple"
    S._MODEL_CACHE.pop("unique", None)


def test_v313_sft_extras_are_off_by_default_and_set_by_env(monkeypatch):
    monkeypatch.delenv("NEFTUNE_ALPHA", raising=False)
    monkeypatch.delenv("GRADIENT_CHECKPOINTING", raising=False)
    assert JOB._sft_extras() == {} and JOB._gradient_checkpointing() is True
    monkeypatch.setenv("NEFTUNE_ALPHA", "5")
    assert JOB._sft_extras() == {"neftune_noise_alpha": 5.0}
    monkeypatch.setenv("NEFTUNE_ALPHA", "abc")
    assert JOB._sft_extras() == {}
    monkeypatch.setenv("GRADIENT_CHECKPOINTING", "0")
    assert JOB._gradient_checkpointing() is False