From 94389b059dc6e7021a13be222b36f62c85ca8cd5 Mon Sep 17 00:00:00 2001 From: Graffioh Date: Wed, 23 Sep 2026 12:46:48 +0000 Subject: [PATCH 1/3] ci(workflows): queue GPU jobs instead of replacing waiting ones A concurrency group keeps one waiting job by default, so when a second PR's GPU job arrives GitHub cancels the first PR's waiting job. Add `queue: max` (up to 100 waiting, first in first out) to the per-device groups of gpu-tests, gpu-tests-amd and speed-profile. Co-authored-by: Claude Opus 5.5 (1M context) --- .github/workflows/ci.yml | 5 +++++ .github/workflows/speed-profile.yml | 2 ++ 2 files changed, 7 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 876b9ec4d..6e2b1293b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -192,6 +192,9 @@ jobs: concurrency: group: ${{ matrix.concurrency_group }} cancel-in-progress: false + # Without a queue GitHub keeps one waiting job per group and cancels the + # older one, so a second PR would silently drop the first PR's GPU run. + queue: max steps: - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 with: @@ -317,6 +320,8 @@ jobs: concurrency: group: lucebox3-${{ matrix.concurrency_key }}-runner cancel-in-progress: false + # Queue instead of replacing waiting jobs (see gpu-tests). + queue: max steps: - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 diff --git a/.github/workflows/speed-profile.yml b/.github/workflows/speed-profile.yml index 5c41b6968..c5349d5d0 100644 --- a/.github/workflows/speed-profile.yml +++ b/.github/workflows/speed-profile.yml @@ -25,6 +25,8 @@ on: concurrency: group: lucebox-rtx3090-gpu-runner cancel-in-progress: false + # Keep every waiting run instead of replacing the older one. + queue: max jobs: speed-profile: From 583deb5d407f409fc397ea9e52da76eda11989d3 Mon Sep 17 00:00:00 2001 From: Graffioh Date: Fri, 25 Sep 2026 06:18:05 +0000 Subject: [PATCH 2/3] ci(e2e): run Qwen and DS4 end to end on lucebox3 Add a Model e2e workflow that runs a real model on lucebox3 and checks whether a change crashes it or changes its behaviour. Qwen3.8-27B (with its dflash2 draft) runs on the R9700 (gfx1201, HIP index 0) and DeepSeek V4 Flash (STRIX mix, sparse prefill) on the Strix Halo (gfx1151, HIP index 1). Each job builds luce_server for its GPU, loads the model, sends 16 fixed prompts with greedy decoding, stops the server, and compares with the last good run on main. It runs only when a maintainer adds the `e2e` label, nightly on main to refresh the baseline, and on manual dispatch. The job fails on a crash, hang, GPU error in the kernel log, a server that will not load or stop, a failed request, a suite that runs out of time, or two or more checks that passed on the baseline and now fail; changed text, a single regressed check, slower decode and a baseline from another ROCm only warn. Baseline runs on main upload their result as the artifact model-e2e-baseline--, and find_baseline.py gives each job the newest one, counting only scheduled or dispatched runs on this repository's main. Each model's GPU, HIP index, model files and server flags sit in one table in select_models.py, and the job reads them from its matrix, with the models under vars.LUCEBOX_MODELS_DIR (default /opt/models). gpu_wait.sh reads the kernel's KFD process list, which any user can read, and fails when it cannot; the job waits up to 4 minutes for other GPU users, then skips with a warning. The KFD health probe moves to .github/ci/kfd_health.sh so both ROCm jobs share it. The job is capped at 35 minutes, with per-step caps, and remembers a clean pass by tree hash, configuration, model files, ROCm version and the baseline's digest. Co-authored-by: Claude Opus 5.5 (1M context) --- .github/ci/e2e/README.md | 46 +++ .github/ci/e2e/find_baseline.py | 98 +++++ .github/ci/e2e/gpu_wait.sh | 48 +++ .github/ci/e2e/prompts.json | 117 ++++++ .github/ci/e2e/run_model_e2e.py | 669 +++++++++++++++++++++++++++++++ .github/ci/e2e/select_models.py | 126 ++++++ .github/ci/e2e/test_model_e2e.py | 486 ++++++++++++++++++++++ .github/ci/kfd_health.sh | 40 ++ .github/workflows/ci.yml | 33 +- .github/workflows/model-e2e.yml | 307 ++++++++++++++ 10 files changed, 1941 insertions(+), 29 deletions(-) create mode 100644 .github/ci/e2e/README.md create mode 100644 .github/ci/e2e/find_baseline.py create mode 100644 .github/ci/e2e/gpu_wait.sh create mode 100644 .github/ci/e2e/prompts.json create mode 100644 .github/ci/e2e/run_model_e2e.py create mode 100644 .github/ci/e2e/select_models.py create mode 100644 .github/ci/e2e/test_model_e2e.py create mode 100644 .github/ci/kfd_health.sh create mode 100644 .github/workflows/model-e2e.yml diff --git a/.github/ci/e2e/README.md b/.github/ci/e2e/README.md new file mode 100644 index 000000000..0ca641227 --- /dev/null +++ b/.github/ci/e2e/README.md @@ -0,0 +1,46 @@ +# Model e2e on lucebox3 + +`model-e2e.yml` runs on lucebox3's self-hosted runner: Qwen3.8-27B on the R9700 +(gfx1201, HIP index 0) and DeepSeek V4 Flash on the Strix Halo (gfx1151, HIP +index 1). Each model's files, flags and GPU are in `select_models.py`. + +## What lucebox3 needs + +- **The models** in `/opt/models` (or the repository variable + `LUCEBOX_MODELS_DIR`): + - `Qwen3.8-27B-UD-IQ4_XS.gguf` + - `qwen38-dflash2-q8_0.gguf` (the draft) + - `DeepSeek-V4-Flash-0731-ROCMFPX-MIX-STRIX.gguf` + + A missing file skips a PR's job and fails a baseline run. +- **A readable kernel log**, so GPU faults during the run are caught: passwordless + `sudo dmesg` for the runner user (as `gpu-tests-amd` already uses), or + `kernel.dmesg_restrict=0`. Without it every run warns that GPU errors were not + checked. +- **ccache** (optional): cold builds take ~100 s without it. The build directory + and remembered passes live in the runner user's `~/.cache/lucebox-e2e`. + +The job sees every user's GPU processes through `/sys/class/kfd/kfd/proc`, which +any user can read. + +## Baselines + +The nightly run on main uploads each model's result as the artifact +`model-e2e-baseline--` (kept 90 days) unless it fails. Every job +compares with the newest one. To refresh a baseline by hand, e.g. after a change +that is meant to alter the output, run the workflow on main with +`update_baseline` ticked. Until the first baseline exists, jobs still fail on +crashes, hangs and failed checks but cannot detect changed output. + +## Benchmarking by hand on lucebox3 + +A job waits up to 4 minutes for other GPU users, then skips with a warning. To +keep jobs off the machine for longer, stop its runner service +(`sudo ./svc.sh stop` in the runner directory) and start it again when you're +done. This also pauses `gpu-tests-amd`. + +## Running the tests + +```bash +cd .github/ci/e2e && uv run --with pytest --no-project python -m pytest -q +``` diff --git a/.github/ci/e2e/find_baseline.py b/.github/ci/e2e/find_baseline.py new file mode 100644 index 000000000..0d8a633b8 --- /dev/null +++ b/.github/ci/e2e/find_baseline.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Find each model's baseline: the newest baseline artifact from a main run. + +Baseline runs on main (nightly, or dispatched with update_baseline) upload +their result as the artifact `model-e2e-baseline--`. This reads +the select job's matrix on stdin and adds to each entry the `baseline_run` that +holds its baseline, or "" when there is none yet. + +Only artifacts from scheduled or dispatched model-e2e runs on this +repository's main count: pull request code runs in the same workflow and could +upload an artifact with the same name. + + python3 find_baseline.py --repo OWNER/REPO < matrix.json +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import urllib.parse +import urllib.request +from collections.abc import Callable + +WORKFLOW = ".github/workflows/model-e2e.yml" +TRUSTED_EVENTS = ("schedule", "workflow_dispatch") + +Api = Callable[[str], dict] + + +def artifact_name(entry: dict) -> str: + return f"model-e2e-baseline-{entry['model']}-{entry['device']}" + + +def github_api(token: str) -> Api: + def get(path: str) -> dict: + request = urllib.request.Request( + f"https://api.github.com/{path}", + headers={ + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {token}", + "X-GitHub-Api-Version": "2022-11-28", + }, + ) + with urllib.request.urlopen(request, timeout=30) as response: + return json.load(response) + + return get + + +def trusted(run: dict, repo: str) -> bool: + return ( + run.get("event") in TRUSTED_EVENTS + and run.get("head_branch") == "main" + and (run.get("head_repository") or {}).get("full_name") == repo + and (run.get("path") or "").split("@")[0] == WORKFLOW + ) + + +def find_run(api: Api, repo: str, name: str) -> str: + """The newest trusted run that uploaded `name`, or "".""" + query = urllib.parse.urlencode({"name": name, "per_page": 50}) + artifacts = api(f"repos/{repo}/actions/artifacts?{query}").get("artifacts", []) + artifacts = [a for a in artifacts if a.get("name") == name and not a.get("expired")] + artifacts.sort(key=lambda a: a.get("created_at") or "", reverse=True) + for artifact in artifacts: + run_id = (artifact.get("workflow_run") or {}).get("id") + if run_id and trusted(api(f"repos/{repo}/actions/runs/{run_id}"), repo): + return str(run_id) + return "" + + +def add_baselines(matrix: dict, api: Api, repo: str) -> dict: + found: dict[str, str] = {} + for entry in matrix["include"]: + name = artifact_name(entry) + if name not in found: + found[name] = find_run(api, repo, name) + entry["baseline_run"] = found[name] + return matrix + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("--repo", required=True, help="OWNER/REPO") + args = parser.parse_args(argv) + token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN") or "" + matrix = add_baselines(json.load(sys.stdin), github_api(token), args.repo) + for entry in matrix["include"]: + where = f"run {entry['baseline_run']}" if entry["baseline_run"] else "none yet" + print(f"Baseline for {entry['model']}: {where}", file=sys.stderr) + print(json.dumps(matrix, separators=(",", ":"))) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/ci/e2e/gpu_wait.sh b/.github/ci/e2e/gpu_wait.sh new file mode 100644 index 000000000..d14933282 --- /dev/null +++ b/.github/ci/e2e/gpu_wait.sh @@ -0,0 +1,48 @@ +#!/usr/bin/env bash +# Wait until no process holds an AMD GPU, so the e2e run does not share the +# machine with someone's manual server or benchmark. +# +# Any KFD process counts: /dev/kfd is shared by every AMD GPU, and a model run +# on the other GPU still competes for host memory and CPU. The holders come +# from the kernel's KFD process list, /sys/class/kfd/kfd/proc/, which +# lists every user's processes and which any user can read. +# +# The last line printed is state=free|busy. Exits 1 when the process list +# cannot be read, since a busy machine would then look free. +# +# Usage: gpu_wait.sh [max seconds to wait, default 300] +set -u + +deadline=$((SECONDS + ${1:-300})) +# Overridable for the tests. +kfd_procs=${KFD_PROC_DIR:-/sys/class/kfd/kfd/proc} + +if [ ! -r "$kfd_procs" ] || [ ! -x "$kfd_procs" ]; then + echo "::error title=GPU check unavailable::Cannot read $kfd_procs to see who holds the GPUs" + exit 1 +fi + +while :; do + # An entry outlives its process for a moment while the driver frees the GPU + # memory; count it until it is gone. + pids=$(ls -A "$kfd_procs") + if [ -z "$pids" ]; then + echo "No process holds the GPUs; they are free." + echo "state=free" + exit 0 + fi + if [ "$SECONDS" -ge "$deadline" ]; then + echo "::warning title=GPU busy::Other processes still hold the GPUs; skipping the model e2e run." + for pid in $pids; do + if ! line=$(ps -o pid=,user=,etime=,args= -p "$pid"); then + line="$pid (exited, but the GPU driver still holds its state)" + fi + echo "${line:0:200}" + done + echo "state=busy" + exit 0 + fi + echo "GPU busy (PIDs: $(echo "$pids" | tr '\n' ' ')); waiting..." + left=$((deadline - SECONDS)) + sleep $((left < 20 ? left : 20)) +done diff --git a/.github/ci/e2e/prompts.json b/.github/ci/e2e/prompts.json new file mode 100644 index 000000000..1557352a6 --- /dev/null +++ b/.github/ci/e2e/prompts.json @@ -0,0 +1,117 @@ +[ + { + "id": "arith", + "messages": [{"role": "user", "content": "What is 17 * 23? Reply with only the number."}], + "max_tokens": 16, + "check": {"number": 391} + }, + { + "id": "capital", + "messages": [{"role": "user", "content": "What is the capital of France? Reply with one word."}], + "max_tokens": 16, + "check": {"contains": ["paris"]} + }, + { + "id": "primes", + "messages": [{"role": "user", "content": "List the first five prime numbers, separated by commas. Reply with only the list."}], + "max_tokens": 32, + "check": {"regex": "2\\D+3\\D+5\\D+7\\D+11"} + }, + { + "id": "word_problem", + "messages": [{"role": "user", "content": "A train travels 60 km in 1.5 hours. What is its average speed in km/h? Reply with only the number."}], + "max_tokens": 16, + "check": {"number": 40} + }, + { + "id": "count", + "messages": [{"role": "user", "content": "Count from 1 to 40, separated by single spaces. Reply with only the numbers."}], + "max_tokens": 160, + "check": {"sequence": 40} + }, + { + "id": "code", + "messages": [{"role": "user", "content": "Write a Python function is_even(n) that returns True when n is even. Reply with only the code."}], + "max_tokens": 96, + "check": {"contains": ["def is_even", "% 2"]} + }, + { + "id": "json", + "messages": [{"role": "user", "content": "Return a JSON object with the key \"name\" set to \"lucebox\" and the key \"version\" set to 3. Reply with only the JSON."}], + "max_tokens": 48, + "check": {"json": {"name": "lucebox", "version": 3}} + }, + { + "id": "translate", + "messages": [{"role": "user", "content": "Translate \"good morning\" into Italian. Reply with only the translation."}], + "max_tokens": 16, + "check": {"any": ["buongiorno", "buon giorno"]} + }, + { + "id": "unicode", + "messages": [{"role": "user", "content": "Repeat this text exactly: CittΓ  東京 πŸš€"}], + "max_tokens": 24, + "check": {"contains": ["cittΓ ", "東京", "πŸš€"]} + }, + { + "id": "multi_turn", + "messages": [ + {"role": "system", "content": "You are a concise assistant."}, + {"role": "user", "content": "My name is Alice and I live in Lisbon."}, + {"role": "assistant", "content": "Nice to meet you, Alice."}, + {"role": "user", "content": "In which city do I live? Reply with one word."} + ], + "max_tokens": 16, + "check": {"contains": ["lisbon"]} + }, + { + "id": "tool_call", + "messages": [{"role": "user", "content": "What is the weather in Rome right now?"}], + "tools": [ + { + "type": "function", + "function": { + "name": "get_weather", + "description": "Get the current weather for a city.", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"] + } + } + } + ], + "max_tokens": 160, + "check": {"tool_call": {"name": "get_weather", "arguments_contain": {"city": "rome"}}} + }, + { + "id": "needle", + "generate": "needle", + "max_tokens": 16, + "check": {"contains": ["7429"]} + }, + { + "id": "stream", + "stream": true, + "messages": [{"role": "user", "content": "What is the capital of Japan? Reply with one word."}], + "max_tokens": 16, + "check": {"contains": ["tokyo"]} + }, + { + "id": "thinking", + "thinking": true, + "messages": [{"role": "user", "content": "What is 12 + 30?"}], + "max_tokens": 768, + "check": {"number": 42, "field": "any"} + }, + { + "id": "story", + "messages": [{"role": "user", "content": "Write a short paragraph about a lighthouse keeper."}], + "max_tokens": 192, + "check": {"min_words": 40, "no_loop": true} + }, + { + "id": "arith_repeat", + "repeat_of": "arith" + } +] diff --git a/.github/ci/e2e/run_model_e2e.py b/.github/ci/e2e/run_model_e2e.py new file mode 100644 index 000000000..6f4a603a0 --- /dev/null +++ b/.github/ci/e2e/run_model_e2e.py @@ -0,0 +1,669 @@ +#!/usr/bin/env python3 +"""Model-backed end-to-end check of luce_server on one GPU. + +Starts the server, waits for the model to load, sends the fixed prompt suite in +prompts.json with greedy decoding, stops the server, and compares the results +with the last good run on main (the baseline). + +Verdict: + fail the server crashed, hung, did not stop, logged GPU errors, never + loaded, a request failed or timed out, the suite ran out of time, two + or more checks that passed on the baseline now fail, or fewer than + half of the checks pass; + warn output text differs from the baseline, one check regressed, a check + fails that also failed on the baseline (or there is no baseline), a + repeated request gave a different answer, decode speed dropped more + than 15%, loading got 50% slower, or the kernel log could not be read + or lost messages during the run; + pass otherwise. + +Text differences and a single regressed check only warn: kernel changes +legitimately flip near-tie tokens, and some short answers sit on a near tie (on +the Strix Halo, DS4 answers 17 * 23 correctly with exact prefill and not with +sparse prefill). A real bug usually breaks several checks at once. +The exit status is 1 on fail and 0 otherwise. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import shlex +import shutil +import signal +import socket +import statistics +import subprocess +import sys +import time +import urllib.request +from collections import Counter +from pathlib import Path + +HERE = Path(__file__).resolve().parent +MIN_PASS_RATE = 0.5 +# Checks that passed on the baseline and now fail; fewer than this only warn. +REGRESSIONS_TO_FAIL = 2 +PERF_DROP = 0.15 +LOAD_GROWTH = 0.5 +# Prompts with at least this many generated tokens feed the decode-speed median. +SPEED_MIN_TOKENS = 32 +GPU_ERROR = re.compile( + r"amdgpu.*(fault|timeout|reset|hang)|kfd.*(fault|error)|ring \S+ timeout", re.IGNORECASE +) +# dmesg's "[seconds since boot]" prefix. +KERNEL_TIME = re.compile(r"^\[\s*(\d+\.\d+)\]") + + +# ─── Prompts and checks ─────────────────────────────────────────────── + + +def needle_messages() -> list[dict]: + colors = ["red", "blue", "green", "amber", "silver", "violet"] + things = ["crate", "ladder", "lantern", "toolbox", "barrel", "rope"] + lines = [] + for i in range(180): + lines.append( + f"Note {i}: the {colors[i % 6]} {things[(i * 5) % 6]} was moved to shelf {(i * 7) % 97}." + ) + if i == 110: + lines.append("Remember this: the secret code is 7429.") + content = ( + "Read these warehouse notes.\n\n" + + "\n".join(lines) + + "\n\nWhat is the secret code mentioned in the notes? Reply with only the code." + ) + return [{"role": "user", "content": content}] + + +def load_prompts(path: Path) -> list[dict]: + prompts = json.loads(path.read_text()) + for prompt in prompts: + # chat() reassembles streamed text only, not streamed tool calls. + if prompt.get("stream") and prompt.get("tools"): + raise ValueError(f"prompt {prompt['id']}: streaming with tools is not supported") + if prompt.get("generate") == "needle": + prompt["messages"] = needle_messages() + return prompts + + +def request_body(prompt: dict) -> dict: + body = { + "model": "luce", + "messages": prompt["messages"], + "max_tokens": prompt["max_tokens"], + "temperature": 0, + "stream": bool(prompt.get("stream")), + "chat_template_kwargs": {"enable_thinking": bool(prompt.get("thinking"))}, + } + if prompt.get("tools"): + body["tools"] = prompt["tools"] + return body + + +def observed_text(message: dict) -> str: + parts = [] + if message.get("reasoning_content"): + parts.append(f"[reasoning] {message['reasoning_content']}") + if message.get("content"): + parts.append(message["content"]) + for call in message.get("tool_calls") or []: + fn = call.get("function") or {} + parts.append(f"[tool] {fn.get('name')} {fn.get('arguments')}") + return "\n".join(parts) + + +def first_json_object(text: str) -> object: + start, end = text.find("{"), text.rfind("}") + if start < 0 or end <= start: + raise ValueError("no JSON object") + return json.loads(text[start : end + 1]) + + +def run_check(check: dict, message: dict) -> tuple[bool, str]: + content = message.get("content") or "" + if check.get("field") == "any": + text = f"{message.get('reasoning_content') or ''}\n{content}" + else: + text = content + lowered = text.lower() + problems = [] + + if "contains" in check: + missing = [s for s in check["contains"] if s.lower() not in lowered] + if missing: + problems.append(f"missing {missing}") + if "any" in check and not any(s.lower() in lowered for s in check["any"]): + problems.append(f"none of {check['any']}") + if "regex" in check and not re.search(check["regex"], text, re.IGNORECASE | re.DOTALL): + problems.append(f"no match for /{check['regex']}/") + if "number" in check: + numbers = [] + for raw in re.findall(r"-?\d[\d,]*(?:\.\d+)?", text): + try: + numbers.append(float(raw.replace(",", ""))) + except ValueError: + continue + if float(check["number"]) not in numbers: + problems.append(f"no {check['number']}") + if "sequence" in check: + want = list(range(1, check["sequence"] + 1)) + if [int(n) for n in re.findall(r"\d+", text)][: len(want)] != want: + problems.append(f"not 1..{check['sequence']}") + if "json" in check: + try: + data = first_json_object(text) + if not isinstance(data, dict) or any( + data.get(k) != v for k, v in check["json"].items() + ): + problems.append(f"JSON {data!r} does not match") + except ValueError as exc: + problems.append(f"invalid JSON ({exc})") + if "tool_call" in check: + want = check["tool_call"] + found = False + for call in message.get("tool_calls") or []: + fn = call.get("function") or {} + arguments = fn.get("arguments") or {} + if isinstance(arguments, str): + try: + arguments = json.loads(arguments) + except ValueError: + arguments = {} + if fn.get("name") == want["name"] and all( + str(v).lower() in str(arguments.get(k, "")).lower() + for k, v in want.get("arguments_contain", {}).items() + ): + found = True + if not found: + problems.append(f"no {want['name']} tool call") + if "min_words" in check and len(text.split()) < check["min_words"]: + problems.append(f"fewer than {check['min_words']} words") + if check.get("no_loop"): + words = text.split() + grams = Counter(tuple(words[i : i + 4]) for i in range(len(words) - 3)) + if grams and max(grams.values()) > 4: + problems.append("repeats itself") + return (not problems, "; ".join(problems)) + + +# ─── HTTP ───────────────────────────────────────────────────────────── + + +def http_json(url: str, body: dict | None, timeout: float) -> dict: + data = json.dumps(body).encode() if body is not None else None + request = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.load(response) + + +def chat(base_url: str, body: dict, timeout: float) -> tuple[dict, dict]: + """Send one chat request; return (message, usage). Streams are reassembled.""" + url = f"{base_url}/v1/chat/completions" + if not body.get("stream"): + reply = http_json(url, body, timeout) + return reply["choices"][0]["message"], reply.get("usage") or {} + + request = urllib.request.Request( + url, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"} + ) + deadline = time.monotonic() + timeout + message: dict = {"content": "", "reasoning_content": ""} + usage: dict = {} + done = False + with urllib.request.urlopen(request, timeout=timeout) as response: + for raw in response: + if time.monotonic() > deadline: + raise TimeoutError("stream exceeded the request timeout") + line = raw.decode().strip() + if not line.startswith("data:"): + continue + payload = line[5:].strip() + if payload == "[DONE]": + done = True + break + chunk = json.loads(payload) + usage = chunk.get("usage") or usage + for choice in chunk.get("choices") or []: + delta = choice.get("delta") or {} + message["content"] += delta.get("content") or "" + message["reasoning_content"] += delta.get("reasoning_content") or "" + if not done: + raise RuntimeError("stream ended without [DONE]") + return message, usage + + +# ─── Server lifecycle ───────────────────────────────────────────────── + + +def rocm_version() -> str | None: + try: + return Path("/opt/rocm/.info/version").read_text().strip() or None + except OSError: + return None + + +def read_kernel_log() -> list[str] | None: + for cmd in (["sudo", "-n", "dmesg"], ["dmesg"]): + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=15) + except (OSError, subprocess.TimeoutExpired): + continue + if proc.returncode == 0: + return proc.stdout.splitlines() + return None + + +def stamped(lines: list[str]) -> list[tuple[float | None, str]] | None: + """Pair each line with its timestamp; continuation lines take the one before. + + None when the log has lines but no timestamps (printk.time=0).""" + out: list[tuple[float | None, str]] = [] + now = None + for line in lines: + match = KERNEL_TIME.match(line) + if match: + now = float(match.group(1)) + out.append((now, line)) + return out if now is not None or not lines else None + + +def kernel_lines_since(before: list[str], after: list[str]) -> tuple[list[str], bool] | None: + """The lines of `after` logged after `before` was read, and whether none can be missing. + + Lines are matched by timestamp, not position: the kernel's ring buffer drops + its oldest lines as new ones arrive, so line counts shift. When `after` no + longer reaches back to the last line of `before`, the buffer wrapped or was + cleared during the run and messages in between may be lost. None when the + log has no timestamps to compare. + """ + old, new = stamped(before), stamped(after) + if old is None or new is None: + return None + if not old: + return after, True + last = old[-1][0] + first = next((t for t, _ in new if t is not None), None) + continuous = first is not None and first <= last + # Lines stamped exactly `last` may be old or new; the old ones are known. + old_at_last = Counter(line for t, line in old if t == last) + lines = [] + for t, line in new: + if t is None or t < last: + continue + if t == last and old_at_last[line] > 0: + old_at_last[line] -= 1 + continue + lines.append(line) + return lines, continuous + + +def stop_server(proc: subprocess.Popen) -> bool: + """Stop the server's process group. Returns False if it would not die.""" + if proc.poll() is not None: + return True + for sig, wait in ((signal.SIGTERM, 30), (signal.SIGKILL, 15)): + try: + os.killpg(proc.pid, sig) + except ProcessLookupError: + return True + try: + proc.wait(timeout=wait) + return True + except subprocess.TimeoutExpired: + continue + return False + + +def run_suite(args: argparse.Namespace, prompts: list[dict], out_dir: Path) -> dict: + cmd = [args.server, args.target] + if args.draft: + cmd += ["--draft", args.draft] + cmd += ["--host", "127.0.0.1", "--port", str(args.port), *shlex.split(args.server_args)] + base_url = f"http://127.0.0.1:{args.port}" + result: dict = { + "model": args.model, + "device": args.device, + "host": socket.gethostname(), + "rocm": rocm_version(), + "commit": args.commit, + "config": { + "target": Path(args.target).name, + "draft": Path(args.draft).name if args.draft else None, + "server_args": args.server_args, + }, + "command": shlex.join(cmd), + "load_seconds": None, + "server": {"crashed": False, "returncode": None, "stuck": False, "loaded": False}, + "budget": args.budget, + "over_budget": 0, # prompts not sent because the suite ran out of time + "gpu_errors": [], + # checked, incomplete (messages may be lost) or unreadable (or no timestamps). + "kernel_log": "unreadable", + "prompts": [], + } + + kernel_before = read_kernel_log() + log = (out_dir / "server.log").open("wb") + started = time.monotonic() + proc = subprocess.Popen(cmd, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + try: + # Load: wait for /health, then a tiny request forces the (lazy) model load. + load_deadline = started + args.load_timeout + while time.monotonic() < load_deadline and proc.poll() is None: + try: + with urllib.request.urlopen(f"{base_url}/health", timeout=5): + break + except OSError: + time.sleep(0.5) + if proc.poll() is None and time.monotonic() < load_deadline: + try: + warmup = {"messages": [{"role": "user", "content": "Say hello."}], "max_tokens": 8} + chat(base_url, {**warmup, "temperature": 0}, load_deadline - time.monotonic()) + result["server"]["loaded"] = True + result["load_seconds"] = round(time.monotonic() - started, 1) + except Exception as exc: + result["load_error"] = f"{type(exc).__name__}: {exc}" + + by_id: dict[str, dict] = {} + budget_deadline = time.monotonic() + args.budget + for prompt in prompts: + source = by_id.get(prompt["repeat_of"]) if prompt.get("repeat_of") else None + spec = source["prompt"] if source else prompt + entry = { + "id": prompt["id"], + "kind": "determinism" if prompt.get("repeat_of") else "check", + "status": "error", + "detail": "", + "observed": "", + "completion_tokens": None, + "seconds": None, + } + result["prompts"].append(entry) + # Follow-on skips: the load or crash failure is reported once, not per prompt. + if not result["server"]["loaded"]: + entry.update(status="skipped", detail="server did not load") + continue + if proc.poll() is not None: + entry.update(status="skipped", detail="server is not running") + continue + if time.monotonic() > budget_deadline: + entry.update(status="skipped", detail="suite budget used up") + result["over_budget"] += 1 + continue + if prompt.get("repeat_of") and source is None: + entry.update(status="skipped", detail=f"{prompt['repeat_of']} got no answer") + continue + t0 = time.monotonic() + # A request never outlasts the suite budget, so the budget bounds the step. + timeout = min(args.request_timeout, budget_deadline - t0) + try: + message, usage = chat(base_url, request_body(spec), timeout) + except Exception as exc: + if time.monotonic() >= budget_deadline: + entry.update(status="skipped", detail="suite budget used up") + result["over_budget"] += 1 + continue + entry["detail"] = f"{type(exc).__name__}: {exc}"[:300] + # A failed request often means the server just died; let it finish + # exiting so the next prompts are skipped rather than failing too. + try: + proc.wait(timeout=2) + except subprocess.TimeoutExpired: + pass + continue + seconds = time.monotonic() - t0 + entry["seconds"] = round(seconds, 2) + entry["completion_tokens"] = usage.get("completion_tokens") + entry["observed"] = observed_text(message) + if prompt.get("repeat_of"): + same = entry["observed"] == source["entry"]["observed"] + entry["status"] = "pass" if same else "differs" + entry["detail"] = "" if same else f"differs from {prompt['repeat_of']}" + else: + ok, detail = run_check(prompt["check"], message) + entry["status"] = "pass" if ok else "fail" + entry["detail"] = detail + by_id[prompt["id"]] = {"prompt": prompt, "entry": entry} + finally: + if proc.poll() is not None: + result["server"]["crashed"] = True + result["server"]["returncode"] = proc.returncode + result["server"]["stuck"] = not stop_server(proc) + log.close() + + kernel_after = read_kernel_log() + since = None + if kernel_before is not None and kernel_after is not None: + since = kernel_lines_since(kernel_before, kernel_after) + if since is not None: + new, continuous = since + result["kernel_log"] = "checked" if continuous else "incomplete" + result["gpu_errors"] = [line for line in new if GPU_ERROR.search(line)][:20] + return result + + +# ─── Evaluation and report ──────────────────────────────────────────── + + +def first_divergence(a: str, b: str) -> int: + for i, (x, y) in enumerate(zip(a, b, strict=False)): + if x != y: + return i + return min(len(a), len(b)) + + +def decode_speed(result: dict) -> float | None: + rates = [ + p["completion_tokens"] / p["seconds"] + for p in result["prompts"] + if p["kind"] == "check" + and p["status"] in ("pass", "fail") + and (p["completion_tokens"] or 0) >= SPEED_MIN_TOKENS + and p["seconds"] + ] + return statistics.median(rates) if rates else None + + +def evaluate(result: dict, baseline: dict | None) -> tuple[list[str], list[str]]: + failures: list[str] = [] + warnings: list[str] = [] + server = result["server"] + if not server["loaded"]: + failures.append( + f"the server did not finish loading ({result.get('load_error', 'no reply')})" + ) + if server["crashed"]: + failures.append(f"the server exited during the run (exit code {server['returncode']})") + if server["stuck"]: + failures.append("the server did not stop after SIGKILL; the GPU driver may be wedged") + if result["gpu_errors"]: + failures.append(f"{len(result['gpu_errors'])} GPU error(s) in the kernel log") + if result.get("over_budget"): + failures.append( + f"the suite used up its {result['budget']:.0f}s budget; " + f"{result['over_budget']} prompt(s) did not run" + ) + kernel_log = result.get("kernel_log", "unreadable") + if kernel_log == "incomplete": + warnings.append( + "the kernel log wrapped or was cleared during the run; GPU errors may be missing" + ) + elif kernel_log == "unreadable": + warnings.append("the kernel log could not be read or compared; GPU errors were not checked") + + base_prompts = {p["id"]: p for p in (baseline or {}).get("prompts", [])} + checks = [p for p in result["prompts"] if p["kind"] == "check" and p["status"] != "skipped"] + regressions = [] + for p in result["prompts"]: + before = base_prompts.get(p["id"]) + if p["status"] == "error": + failures.append(f"`{p['id']}`: request failed: {p['detail']}") + elif p["status"] == "fail": + if before and before["status"] == "pass": + regressions.append(f"`{p['id']}`: passed on the baseline, now fails: {p['detail']}") + else: + warnings.append(f"`{p['id']}`: check fails: {p['detail']}") + elif p["status"] == "differs": + warnings.append(f"`{p['id']}`: the same request gave a different answer") + if ( + before + and p["kind"] == "check" + and p["status"] in ("pass", "fail") + and before["observed"] != p["observed"] + ): + at = first_divergence(before["observed"], p["observed"]) + warnings.append(f"`{p['id']}`: output differs from the baseline from character {at}") + + (failures if len(regressions) >= REGRESSIONS_TO_FAIL else warnings).extend(regressions) + + passed = sum(p["status"] == "pass" for p in checks) + if checks and server["loaded"] and passed / len(checks) < MIN_PASS_RATE: + failures.append(f"only {passed}/{len(checks)} checks pass") + + if baseline: + if baseline.get("config") != result.get("config"): + warnings.append("the baseline used a different model or server configuration") + # A ROCm upgrade since the baseline explains drift. + if baseline.get("rocm") != result.get("rocm"): + warnings.append( + f"the baseline ran on ROCm {baseline.get('rocm') or '?'}, " + f"this run on {result.get('rocm') or '?'}" + ) + now, then = decode_speed(result), decode_speed(baseline) + if now and then and now < then * (1 - PERF_DROP): + warnings.append(f"decode speed {now:.1f} tok/s vs {then:.1f} on the baseline") + load_now, load_then = result.get("load_seconds"), baseline.get("load_seconds") + if load_now and load_then and load_now > load_then * (1 + LOAD_GROWTH): + warnings.append(f"load took {load_now:.0f}s vs {load_then:.0f}s on the baseline") + return failures, warnings + + +def render_report(result: dict, baseline: dict | None, title: str, log_tail: str) -> str: + icon = {"pass": "βœ…", "warn": "⚠️", "fail": "❌"}[result["verdict"]] + base_prompts = {p["id"]: p for p in (baseline or {}).get("prompts", [])} + speed = decode_speed(result) + lines = [ + f"## {icon} {title}: {result['verdict']}", + "", + f"- Model: `{result['config']['target']}`" + + (f" + draft `{result['config']['draft']}`" if result["config"]["draft"] else ""), + f"- Server args: `{result['config']['server_args'] or '(none)'}`", + f"- ROCm: {result.get('rocm') or 'unknown'}", + f"- Load: {result['load_seconds']}s Β· decode median: " + + (f"{speed:.1f} tok/s" if speed else "n/a"), + "- Baseline: " + + ( + f"main @ `{(baseline.get('commit') or '?')[:10]}` on {baseline.get('host', '?')}" + if baseline + else "none yet" + ), + "- Kernel log: " + + { + "checked": "checked", + "incomplete": "checked, but messages were lost during the run", + "unreadable": "not readable (or has no timestamps)", + }[result["kernel_log"]], + "", + ] + if result["failures"]: + lines += ["**Failures**", "", *[f"- {f}" for f in result["failures"]], ""] + if result["warnings"]: + lines += ["**Warnings**", "", *[f"- {w}" for w in result["warnings"]], ""] + lines += ["| Prompt | Result | Tokens | Seconds | vs baseline |", "|---|---|---:|---:|---|"] + for p in result["prompts"]: + before = base_prompts.get(p["id"]) + if not before: + versus = "" + elif before["observed"] == p["observed"]: + versus = "same text" + else: + versus = f"differs @ {first_divergence(before['observed'], p['observed'])}" + detail = f" ({p['detail']})" if p["detail"] else "" + lines.append( + f"| `{p['id']}` | {p['status']}{detail} | {p['completion_tokens'] or ''} " + f"| {p['seconds'] or ''} | {versus} |" + ) + if result["gpu_errors"]: + lines += ["", "**GPU errors in the kernel log**", "", "```", *result["gpu_errors"], "```"] + if result["verdict"] == "fail" and log_tail: + lines += [ + "", + "
Server log (tail)", + "", + "```", + log_tail, + "```", + "", + "
", + ] + return "\n".join(lines) + "\n" + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument("--model", required=True, help="label, e.g. qwen or ds4") + parser.add_argument("--device", required=True, help="label, e.g. r9700") + parser.add_argument("--server", required=True, help="luce_server binary") + parser.add_argument("--target", required=True, help="target GGUF") + parser.add_argument("--draft", help="draft GGUF") + parser.add_argument("--server-args", default="", help="extra server flags, shell-quoted") + parser.add_argument("--port", type=int, default=18200) + parser.add_argument("--out-dir", required=True) + parser.add_argument("--prompts", default=str(HERE / "prompts.json")) + parser.add_argument("--baseline", help="result JSON of the last good main run") + parser.add_argument("--write-baseline", help="store this run here unless it fails") + parser.add_argument("--commit", default=os.environ.get("GITHUB_SHA", "")) + parser.add_argument("--load-timeout", type=float, default=420) + parser.add_argument("--request-timeout", type=float, default=120) + parser.add_argument("--budget", type=float, default=600, help="seconds for the whole suite") + args = parser.parse_args(argv) + + out_dir = Path(args.out_dir) + out_dir.mkdir(parents=True, exist_ok=True) + if not shutil.which(args.server): + print(f"::error::server binary not found: {args.server}") + return 2 + for label, path in (("target", args.target), ("draft", args.draft)): + if path and not Path(path).is_file(): + print(f"::error::{label} model not found: {path}") + return 2 + + baseline = None + if args.baseline and Path(args.baseline).is_file(): + baseline = json.loads(Path(args.baseline).read_text()) + + result = run_suite(args, load_prompts(Path(args.prompts)), out_dir) + failures, warnings = evaluate(result, baseline) + result["failures"], result["warnings"] = failures, warnings + result["verdict"] = "fail" if failures else "warn" if warnings else "pass" + + log_lines = (out_dir / "server.log").read_text(errors="replace").splitlines() + report = render_report( + result, + baseline, + f"{args.model} on {args.device} @ {result['host']}", + "\n".join(log_lines[-60:]), + ) + (out_dir / "result.json").write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n") + (out_dir / "report.md").write_text(report) + print(report) + for failure in failures: + print(f"::error title=Model e2e ({args.model})::{failure}") + for warning in warnings: + print(f"::warning title=Model e2e ({args.model})::{warning}") + + if args.write_baseline and result["verdict"] != "fail": + target = Path(args.write_baseline) + target.parent.mkdir(parents=True, exist_ok=True) + tmp = target.with_suffix(".tmp") + tmp.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n") + tmp.replace(target) + print(f"Baseline updated: {target}") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/ci/e2e/select_models.py b/.github/ci/e2e/select_models.py new file mode 100644 index 000000000..d86bc8653 --- /dev/null +++ b/.github/ci/e2e/select_models.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""Pick which model e2e jobs a change needs and print the GitHub Actions matrix. + +Changed paths come on stdin, one per line. A path under a model's own sources +selects that model; a path in shared server code selects every model; anything +else (docs, scripts, other models) selects nothing. + +Each matrix entry carries everything its job needs: the GPU on lucebox3, the +model files (in the models directory) and the server flags. + + gh api repos/OWNER/REPO/pulls/N/files --paginate --jq '.[].filename' \\ + | python3 .github/ci/e2e/select_models.py --mode paths +""" + +from __future__ import annotations + +import argparse +import json +import sys + +# HIP indices as on lucebox3 (see gpu-tests-amd in ci.yml). +MODELS = { + "qwen": { + "device": "r9700", + "device_name": "Radeon AI PRO R9700", + "arch": "gfx1201", + "hip_index": 0, + "target": "Qwen3.8-27B-UD-IQ4_XS.gguf", + "draft": "qwen38-dflash2-q8_0.gguf", + "server_args": "--max-ctx 8192 --disk-prefix-cache off", + }, + "ds4": { + "device": "strix-halo", + "device_name": "Strix Halo Radeon 8060S", + "arch": "gfx1151", + "hip_index": 1, + "target": "DeepSeek-V4-Flash-0731-ROCMFPX-MIX-STRIX.gguf", + "draft": "", + # Sparse (batched) prefill: exact prefill runs at ~21 tok/s on the Strix + # Halo, too slow for the 2.5K-token needle prompt within the job budget. + "server_args": "--max-ctx 8192 --ds4-fused-decode --ds4-expert-top-k 6" + " --ds4-prefill sparse --prefix-cache-slots 0 --prefill-cache-slots 0" + " --disk-prefix-cache off", + }, +} + +MODEL_PREFIXES = { + "ds4": ("server/src/deepseek4/",), + "qwen": ( + "server/src/qwen35/", + "server/src/qwen3/", + "server/src/draft/", + "server/src/delta_net", + "server/src/flashprefill", + "server/src/pflash_", + ), +} +# Other model families: changes there cannot affect Qwen or DS4. +OTHER_MODEL_PREFIXES = ( + "server/src/bailingmoe3/", + "server/src/gemma4/", + "server/src/laguna/", + "server/src/qwen35moe/", +) +SHARED_PREFIXES = ( + "server/src/", + "server/include/", + "server/deps/", + "server/cmake/", + "server/hip_compat/", + "server/CMakeLists.txt", + ".github/ci/e2e/", + ".github/ci/kfd_health.sh", + ".github/workflows/model-e2e.yml", +) + + +def models_for(paths: list[str]) -> list[str]: + selected: set[str] = set() + for path in paths: + owners = [m for m, prefixes in MODEL_PREFIXES.items() if path.startswith(prefixes)] + if owners: + selected.update(owners) + elif path.startswith(OTHER_MODEL_PREFIXES): + continue + elif path.startswith(SHARED_PREFIXES): + selected.update(MODELS) + return sorted(selected) + + +def matrix(models: list[str]) -> dict: + return {"include": [{"model": model, **MODELS[model]} for model in models]} + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + parser.add_argument( + "--mode", + default="paths", + help="'paths' reads changed paths from stdin, 'all' runs every model, " + "or a comma-separated list of models", + ) + parser.add_argument( + "--fallback-all", + action="store_true", + help="run every model when the paths select none (an explicit request)", + ) + args = parser.parse_args(argv) + + if args.mode == "paths": + models = models_for([line.strip() for line in sys.stdin if line.strip()]) + if not models and args.fallback_all: + models = sorted(MODELS) + elif args.mode == "all": + models = sorted(MODELS) + else: + models = sorted({m.strip() for m in args.mode.split(",") if m.strip()}) + unknown = [m for m in models if m not in MODELS] + if unknown: + parser.error(f"unknown model(s): {', '.join(unknown)}") + print(json.dumps(matrix(models), separators=(",", ":"))) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/ci/e2e/test_model_e2e.py b/.github/ci/e2e/test_model_e2e.py new file mode 100644 index 000000000..db8477965 --- /dev/null +++ b/.github/ci/e2e/test_model_e2e.py @@ -0,0 +1,486 @@ +"""Tests for the model e2e scripts. A small fake server stands in for luce_server.""" + +from __future__ import annotations + +import json +import os +import socket +import subprocess +import sys +import textwrap +import time +from pathlib import Path + +import pytest +import run_model_e2e +from find_baseline import add_baselines +from run_model_e2e import evaluate, kernel_lines_since, load_prompts, main, run_check +from select_models import matrix, models_for + +HERE = Path(__file__).resolve().parent + +# The fake server's answer to each prompt in prompts.json, found by a phrase in +# the prompt's last message (the tool-call prompt about the weather is handled +# separately). test_the_fake_answers_every_prompt keeps this in step with +# prompts.json. +ANSWERS = [ + ("17 * 23", "391"), + ("capital of France", "Paris"), + ("prime numbers", "2, 3, 5, 7, 11"), + ("train travels", "40"), + ("Count from 1 to 40", " ".join(str(i) for i in range(1, 41))), + ("is_even", "def is_even(n):\n return n % 2 == 0"), + ("JSON object", '{"name": "lucebox", "version": 3}'), + ("good morning", "Buongiorno"), + ("Repeat this text", "CittΓ  東京 πŸš€"), + ("which city", "Lisbon."), + ("secret code", "7429"), + ("capital of Japan", "Tokyo"), + ("12 + 30", "42"), + ("lighthouse", " ".join(f"word{i}" for i in range(60))), + ("Say hello", "Hello"), +] + +FAKE_SERVER = textwrap.dedent( + """ + import json, os, sys, time + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + ANSWERS = json.loads(os.environ["FAKE_ANSWERS"]) + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_GET(self): + self.send_response(200) + self.end_headers() + self.wfile.write(b"ok") + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + question = body["messages"][-1]["content"] + crash = os.environ.get("FAKE_CRASH_ON") + slow = os.environ.get("FAKE_SLOW_ON") + if slow and slow in question: + time.sleep(30) + if os.environ.get("FAKE_WRONG") and "train travels" in question: + question = "Say hello" + if crash and crash in question: + sys.stderr.write("fake server: simulated crash\\n") + sys.stderr.flush() + os._exit(3) + message = {"role": "assistant", "content": ""} + if "weather" in question: + message["tool_calls"] = [{"type": "function", "function": { + "name": "get_weather", "arguments": json.dumps({"city": "Rome"})}}] + else: + message["content"] = next(a for q, a in ANSWERS if q in question) + if os.environ.get("FAKE_WRONG") and "17 * 23" in question: + message["content"] = "392" + usage = {"prompt_tokens": 10, "completion_tokens": len(message["content"].split())} + if body.get("stream"): + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.end_headers() + for word in message["content"].split(" "): + chunk = {"choices": [{"delta": {"content": word + " "}}]} + self.wfile.write(f"data: {json.dumps(chunk)}\\n\\n".encode()) + self.wfile.write(b"data: [DONE]\\n\\n") + return + data = json.dumps({"choices": [{"message": message}], "usage": usage}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + port = int(sys.argv[sys.argv.index("--port") + 1]) + ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever() + """ +) + + +def free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.fixture(autouse=True) +def quiet_kernel_log(monkeypatch: pytest.MonkeyPatch) -> None: + # The machine running the tests may not let us read its kernel log. + monkeypatch.setattr(run_model_e2e, "read_kernel_log", lambda: ["[ 1.000000] boot"]) + + +@pytest.fixture +def fake(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.setenv("FAKE_ANSWERS", json.dumps(ANSWERS)) + server = tmp_path / "fake_server.py" + server.write_text(FAKE_SERVER) + return server + + +def test_the_fake_answers_every_prompt() -> None: + # The healthy-run tests send the real prompts.json to the fake server. A new + # or reworded prompt needs an entry in ANSWERS (and an answer that passes + # its check), or those tests fail with a confusing request error. + prompts = load_prompts(HERE / "prompts.json") + for prompt in prompts: + if prompt.get("repeat_of"): + continue + question = prompt["messages"][-1]["content"] + assert "weather" in question or any(q in question for q, _ in ANSWERS), ( + f"prompt {prompt['id']!r} has no answer in ANSWERS" + ) + # test_crash_fails_and_skips_the_rest crashes on `primes`. + ids = [p["id"] for p in prompts] + assert ids.index("capital") < ids.index("primes") < ids.index("count") + + +def run(fake: Path, out: Path, *extra: str) -> tuple[int, dict]: + # The fake is launched as `python fake_server.py `: python is the + # "server binary" and the script sits where the target GGUF would. + code = main([ + "--model", "fake", "--device", "cpu", "--server", sys.executable, + "--target", str(fake), "--port", str(free_port()), "--out-dir", str(out), + "--load-timeout", "20", "--request-timeout", "10", *extra, + ]) # fmt: skip + return code, json.loads((out / "result.json").read_text()) + + +def test_healthy_server_passes_and_writes_a_baseline(fake: Path, tmp_path: Path) -> None: + baseline = tmp_path / "baseline.json" + code, result = run(fake, tmp_path / "a", "--write-baseline", str(baseline)) + assert code == 0, result["failures"] + assert result["verdict"] == "pass" + assert all(p["status"] == "pass" for p in result["prompts"]) + assert baseline.is_file() + + code, again = run(fake, tmp_path / "b", "--baseline", str(baseline)) + assert code == 0 and again["verdict"] == "pass" + assert "same text" in (tmp_path / "b" / "report.md").read_text() + + +def test_regressions_against_the_baseline_fail( + fake: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + baseline = tmp_path / "baseline.json" + run(fake, tmp_path / "a", "--write-baseline", str(baseline)) + monkeypatch.setenv("FAKE_WRONG", "1") + code, result = run(fake, tmp_path / "b", "--baseline", str(baseline)) + assert code == 1 + assert any("`arith`: passed on the baseline" in f for f in result["failures"]) + assert any("`word_problem`: passed on the baseline" in f for f in result["failures"]) + + +def test_a_single_regression_only_warns() -> None: + base = result(prompt("a", "pass", "x"), prompt("b", "pass", "y")) + now = result(prompt("a", "fail", "x2"), prompt("b", "pass", "y")) + failures, warnings = evaluate(now, base) + assert failures == [] + assert any("`a`: passed on the baseline" in w for w in warnings) + + +def test_crash_fails_and_skips_the_rest( + fake: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("FAKE_CRASH_ON", "prime numbers") + code, result = run(fake, tmp_path / "a") + assert code == 1 + assert result["server"]["crashed"] and result["server"]["returncode"] == 3 + statuses = {p["id"]: p["status"] for p in result["prompts"]} + assert statuses["capital"] == "pass" + assert statuses["primes"] == "error" + assert statuses["count"] == "skipped" + assert not any("checks pass" in f for f in result["failures"]) + assert "simulated crash" in (tmp_path / "a" / "report.md").read_text() + + +def test_missing_server_is_a_setup_error(tmp_path: Path) -> None: + code = main([ + "--model", "x", "--device", "y", "--server", str(tmp_path / "luce_server"), + "--target", str(tmp_path), "--out-dir", str(tmp_path), + ]) # fmt: skip + assert code == 2 + + +def test_missing_target_is_a_setup_error(tmp_path: Path) -> None: + code = main([ + "--model", "x", "--device", "y", "--server", "true", + "--target", str(tmp_path / "missing.gguf"), "--out-dir", str(tmp_path), + ]) # fmt: skip + assert code == 2 + + +def test_missing_draft_is_a_setup_error(fake: Path, tmp_path: Path) -> None: + # Running without the draft would test a different configuration. + code = main([ + "--model", "x", "--device", "y", "--server", "true", "--target", str(fake), + "--draft", str(tmp_path / "missing.gguf"), "--out-dir", str(tmp_path), + ]) # fmt: skip + assert code == 2 + + +def test_running_out_of_budget_fails_as_such(fake: Path, tmp_path: Path) -> None: + code, result = run(fake, tmp_path / "a", "--budget", "0") + assert code == 1 + assert result["over_budget"] == len(result["prompts"]) + assert any("used up its 0s budget" in f for f in result["failures"]) + assert not any("request failed" in f for f in result["failures"]) + + +def test_a_slow_request_stops_at_the_budget( + fake: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # The request timeout (10 s) is longer than what is left of the budget. + monkeypatch.setenv("FAKE_SLOW_ON", "prime numbers") + started = time.monotonic() + code, result = run(fake, tmp_path / "a", "--budget", "3") + assert time.monotonic() - started < 10 + assert code == 1 + primes = next(p for p in result["prompts"] if p["id"] == "primes") + assert primes["status"] == "skipped" + assert any("budget" in f for f in result["failures"]) + assert not any("request failed" in f for f in result["failures"]) + + +def test_streaming_prompts_cannot_use_tools(tmp_path: Path) -> None: + prompts = tmp_path / "prompts.json" + prompts.write_text(json.dumps([{"id": "x", "stream": True, "tools": [{}]}])) + with pytest.raises(ValueError, match="streaming with tools"): + load_prompts(prompts) + + +@pytest.mark.parametrize( + ("check", "message", "ok"), + [ + ({"number": 391}, {"content": "The answer is 391."}, True), + ({"number": 391}, {"content": "392"}, False), + ({"sequence": 5}, {"content": "1 2 3 4 5"}, True), + ({"sequence": 5}, {"content": "1 2 4 5"}, False), + ({"json": {"v": 3}}, {"content": '```json\n{"v": 3}\n```'}, True), + ({"json": {"v": 3}}, {"content": "{v: 3}"}, False), + ({"any": ["buongiorno", "buon giorno"]}, {"content": "Buon giorno!"}, True), + ({"number": 42, "field": "any"}, {"reasoning_content": "so 42", "content": ""}, True), + ({"number": 42}, {"reasoning_content": "so 42", "content": ""}, False), + ( + {"tool_call": {"name": "f", "arguments_contain": {"city": "rome"}}}, + {"tool_calls": [{"function": {"name": "f", "arguments": '{"city": "Rome"}'}}]}, + True, + ), + ({"no_loop": True}, {"content": "a b c d " * 6}, False), + ({"min_words": 3}, {"content": "one two"}, False), + ], +) +def test_checks(check: dict, message: dict, ok: bool) -> None: + assert run_check(check, message)[0] is ok + + +def prompt(pid: str, status: str, observed: str, tokens: int = 64, seconds: float = 2.0) -> dict: + return { + "id": pid, "kind": "check", "status": status, "detail": "", "observed": observed, + "completion_tokens": tokens, "seconds": seconds, + } # fmt: skip + + +def result(*prompts: dict, load: float = 10.0) -> dict: + return { + "config": {"target": "m.gguf", "draft": None, "server_args": ""}, + "load_seconds": load, + "server": {"crashed": False, "returncode": None, "stuck": False, "loaded": True}, + "gpu_errors": [], + "kernel_log": "checked", + "prompts": list(prompts), + } + + +def test_evaluate_warns_on_drift_speed_and_load() -> None: + base = result(prompt("a", "pass", "hello world", seconds=2.0), load=10) + now = result(prompt("a", "pass", "hello there", seconds=4.0), load=30) + failures, warnings = evaluate(now, base) + assert failures == [] + assert any("differs from the baseline from character 6" in w for w in warnings) + assert any("decode speed" in w for w in warnings) + assert any("load took" in w for w in warnings) + + +def test_evaluate_only_warns_on_checks_that_already_failed() -> None: + base = result(prompt("a", "fail", "x"), prompt("b", "pass", "y")) + now = result(prompt("a", "fail", "x"), prompt("b", "pass", "y")) + failures, warnings = evaluate(now, base) + assert failures == [] and any("`a`: check fails" in w for w in warnings) + + +def test_evaluate_fails_on_gpu_errors_and_low_pass_rate() -> None: + now = result(prompt("a", "fail", "x"), prompt("b", "fail", "y"), prompt("c", "pass", "z")) + now["gpu_errors"] = ["amdgpu: ring gfx_0.0.0 timeout"] + failures, _ = evaluate(now, None) + assert any("GPU error" in f for f in failures) + assert any("only 1/3 checks pass" in f for f in failures) + + +def test_evaluate_warns_when_the_kernel_log_is_incomplete_or_unreadable() -> None: + now = result(prompt("a", "pass", "x")) + now["kernel_log"] = "incomplete" + assert any("wrapped or was cleared" in w for w in evaluate(now, None)[1]) + now["kernel_log"] = "unreadable" + assert any("could not be read" in w for w in evaluate(now, None)[1]) + + +def test_run_reports_gpu_errors_logged_during_the_run( + fake: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + logs = iter([ + ["[ 1.000000] boot"], + ["[ 1.000000] boot", "[ 99.000000] amdgpu: ring gfx_0.0.0 timeout"], + ]) # fmt: skip + monkeypatch.setattr(run_model_e2e, "read_kernel_log", lambda: next(logs)) + code, result = run(fake, tmp_path / "a") + assert code == 1 + assert result["kernel_log"] == "checked" + assert result["gpu_errors"] == ["[ 99.000000] amdgpu: ring gfx_0.0.0 timeout"] + + +def test_kernel_log_new_lines_survive_ring_buffer_rollover() -> None: + before = [f"[ {i}.000000] old {i}" for i in range(1, 6)] + # Two old lines rolled out while two new ones, one a GPU fault, arrived: + # the line count is unchanged, but the new lines are still found. + after = before[2:] + ["[ 10.000000] new", "[ 11.000000] amdgpu: ring gfx timeout"] + assert kernel_lines_since(before, after) == ( + ["[ 10.000000] new", "[ 11.000000] amdgpu: ring gfx timeout"], + True, + ) + + +def test_kernel_log_reports_lost_continuity() -> None: + before = ["[ 1.000000] a", "[ 2.000000] b"] + # Nothing from before is left: the buffer wrapped or was cleared. + after = ["[ 7.000000] x", "[ 8.000000] y"] + assert kernel_lines_since(before, after) == (after, False) + assert kernel_lines_since(before, []) == ([], False) + + +def test_kernel_log_same_timestamp_and_continuation_lines() -> None: + before = ["[ 1.000000] a", "[ 2.000000] b", " b continued"] + after = [*before, "[ 2.000000] c", " c continued", "[ 3.000000] d"] + assert kernel_lines_since(before, after) == ( + ["[ 2.000000] c", " c continued", "[ 3.000000] d"], + True, + ) + + +def test_kernel_log_without_timestamps_cannot_be_compared() -> None: + assert kernel_lines_since(["a", "b"], ["a", "b", "c"]) is None + assert kernel_lines_since([], ["[ 1.000000] a"]) == (["[ 1.000000] a"], True) + + +def gpu_wait(procs: Path) -> subprocess.CompletedProcess: + """Run gpu_wait.sh with no wait against a fake KFD process list.""" + return subprocess.run( + ["bash", str(HERE / "gpu_wait.sh"), "0"], + env={**os.environ, "KFD_PROC_DIR": str(procs)}, capture_output=True, text=True, timeout=30, + ) # fmt: skip + + +def test_gpu_wait_reads_the_kernel_process_list(tmp_path: Path) -> None: + procs = tmp_path / "procs" + procs.mkdir() + assert gpu_wait(procs).stdout.splitlines()[-1] == "state=free" + # Any user's process counts, even one that is still being torn down. + (procs / "999999999").mkdir() + assert gpu_wait(procs).stdout.splitlines()[-1] == "state=busy" + + +def test_gpu_wait_fails_when_it_cannot_see_every_user(tmp_path: Path) -> None: + assert gpu_wait(tmp_path / "missing").returncode == 1 + procs = tmp_path / "procs" + (procs / "123").mkdir(parents=True) + procs.chmod(0) + try: + if os.access(procs, os.R_OK): + pytest.skip("running as root: permissions are not enforced") + assert gpu_wait(procs).returncode == 1 + finally: + procs.chmod(0o755) + + +@pytest.mark.parametrize( + ("paths", "models"), + [ + (["server/src/deepseek4/ds4_graph.cpp"], ["ds4"]), + (["server/src/qwen35/qwen35_target_graph.cpp"], ["qwen"]), + (["server/src/draft/dflash_draft.cpp", "docs/x.md"], ["qwen"]), + (["server/src/server/http_server.cpp"], ["ds4", "qwen"]), + (["server/deps/llama.cpp/ggml/src/ggml.c"], ["ds4", "qwen"]), + (["server/src/laguna/laguna.cpp", "README.md", "server/scripts/x.py"], []), + ([".github/ci/e2e/prompts.json"], ["ds4", "qwen"]), + ], +) +def test_select_models(paths: list[str], models: list[str]) -> None: + assert models_for(paths) == models + + +def test_matrix_carries_each_models_gpu_and_files() -> None: + entries = {e["model"]: e for e in matrix(["ds4", "qwen"])["include"]} + assert (entries["qwen"]["arch"], entries["qwen"]["hip_index"]) == ("gfx1201", 0) + assert (entries["ds4"]["arch"], entries["ds4"]["hip_index"]) == ("gfx1151", 1) + assert entries["qwen"]["draft"] and not entries["ds4"]["draft"] + assert "--ds4-prefill sparse" in entries["ds4"]["server_args"] + + +REPO = "Luce-Org/lucebox" + + +def fake_api(artifacts: list[dict], runs: dict[int, dict]): + def get(path: str) -> dict: + if "/actions/artifacts?" in path: + name = path.split("name=")[1].split("&")[0] + return {"artifacts": [a for a in artifacts if a["name"] == name]} + return runs[int(path.rsplit("/", 1)[1])] + + return get + + +def main_run(event: str = "schedule", **overrides: object) -> dict: + return { + "event": event, + "head_branch": "main", + "head_repository": {"full_name": REPO}, + "path": ".github/workflows/model-e2e.yml", + **overrides, + } + + +def artifact(run_id: int, created: str, name: str = "model-e2e-baseline-qwen-r9700") -> dict: + return {"name": name, "expired": False, "created_at": created, "workflow_run": {"id": run_id}} + + +def test_baseline_is_the_newest_from_a_trusted_main_run() -> None: + artifacts = [ + artifact(1, "2026-09-01T03:00:00Z"), + artifact(2, "2026-09-03T03:00:00Z"), + # Newer, but uploaded by pull request code or from a fork's main. + artifact(3, "2026-09-04T03:00:00Z"), + artifact(4, "2026-09-05T03:00:00Z"), + artifact(5, "2026-09-06T03:00:00Z"), + {**artifact(6, "2026-09-07T03:00:00Z"), "expired": True}, + ] + runs = { + 1: main_run(), + 2: main_run("workflow_dispatch"), + 3: main_run("pull_request", head_branch="feature"), + 4: main_run(head_repository={"full_name": "someone/lucebox"}), + 5: main_run(path=".github/workflows/ci.yml"), + 6: main_run(), + } + got = add_baselines(matrix(["ds4", "qwen"]), fake_api(artifacts, runs), REPO) + assert {e["model"]: e["baseline_run"] for e in got["include"]} == {"ds4": "", "qwen": "2"} + + +def test_evaluate_warns_when_the_baseline_ran_on_another_rocm() -> None: + base, now = result(prompt("a", "pass", "x")), result(prompt("a", "pass", "x")) + base["rocm"], now["rocm"] = "7.1.0", "7.2.0" + assert any("ROCm 7.1.0" in w for w in evaluate(now, base)[1]) + now["rocm"] = "7.1.0" + assert evaluate(now, base) == ([], []) diff --git a/.github/ci/kfd_health.sh b/.github/ci/kfd_health.sh new file mode 100644 index 000000000..4b17b957d --- /dev/null +++ b/.github/ci/kfd_health.sh @@ -0,0 +1,40 @@ +#!/usr/bin/env bash +# Fail fast, with a diagnosis, when ROCm's KFD driver is wedged. +# +# rocminfo on a wedged KFD blocks in uninterruptible sleep (D-state): timeout(1) +# cannot kill it and a foreground wait blocks until the job timeout. Probe in +# the background (output to a file so no pipe keeps the step alive) and enforce +# the deadline in the shell. On a hang, dump the evidence (D-state holders, +# dmesg) and exit 1 within seconds. Each diagnostic has its own deadline, and +# sudo never prompts: reading a wedged process's state can block too. +# +# Usage: kfd_health.sh [rocminfo output file, default /tmp/rocminfo.out] +set -u + +out="${1:-/tmp/rocminfo.out}" +/opt/rocm/bin/rocminfo > "$out" 2>&1 & +probe=$! +for _ in $(seq 1 15); do + kill -0 "$probe" 2>/dev/null || break + sleep 1 +done +if kill -0 "$probe" 2>/dev/null; then + echo "::error::rocminfo hung (likely D-state) β€” ROCm/KFD wedged; the box needs a reboot" + echo "--- probe state:" + ps -o pid,stat,wchan:32,comm -p "$probe" || true + echo "--- processes holding /dev/kfd:" + timeout -k 2 10 sudo -n fuser -v /dev/kfd 2>&1 || true + echo "--- D-state processes:" + ps -eo pid,user,stat,wchan:32,comm | awk '$3 ~ /D/' || true + echo "--- recent amdgpu/kfd dmesg:" + timeout -k 2 10 sudo -n dmesg 2>/dev/null | grep -iE "amdgpu|kfd" | tail -15 || true + kill -9 "$probe" 2>/dev/null || true + disown "$probe" 2>/dev/null || true + exit 1 +fi +if ! wait "$probe"; then + echo "::error::rocminfo exited non-zero" + tail -5 "$out" + exit 1 +fi +echo "KFD healthy" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6e2b1293b..284f6298e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -327,35 +327,10 @@ jobs: - name: KFD health (diagnose instead of hanging) # rocminfo on a wedged KFD blocks in uninterruptible sleep and eats - # the whole 20-minute job timeout. Probe with a hard timeout first, - # and when it hangs, dump the evidence (D-state holders, dmesg) so - # the job fails in seconds with a diagnosis instead of silently. - run: | - # A wedged KFD puts rocminfo in UNINTERRUPTIBLE sleep: timeout(1) - # cannot kill it and a foreground wait blocks until the job - # timeout. Probe in the background (output to a file so no pipe - # keeps the step alive) and enforce the deadline in the shell. - /opt/rocm/bin/rocminfo > /tmp/rocminfo.out 2>&1 & - PROBE=$! - for i in $(seq 1 15); do - kill -0 $PROBE 2>/dev/null || break - sleep 1 - done - if kill -0 $PROBE 2>/dev/null; then - echo "::error::rocminfo hung (likely D-state) β€” ROCm/KFD wedged; the box needs a reboot" - echo "--- probe state:" - ps -o pid,stat,wchan:32,comm -p $PROBE || true - echo "--- processes holding /dev/kfd:" - sudo fuser -v /dev/kfd 2>&1 || true - echo "--- D-state processes:" - ps -eo pid,user,stat,wchan:32,comm | awk '$3 ~ /D/' || true - echo "--- recent amdgpu/kfd dmesg:" - sudo dmesg 2>/dev/null | grep -iE "amdgpu|kfd" | tail -15 || true - kill -9 $PROBE 2>/dev/null || true - disown $PROBE 2>/dev/null || true - exit 1 - fi - wait $PROBE && echo "KFD healthy" || { echo "::error::rocminfo exited non-zero"; cat /tmp/rocminfo.out | tail -5; exit 1; } + # the whole 20-minute job timeout. The probe enforces a 15 s deadline + # and dumps the evidence (D-state holders, dmesg) so the job fails in + # seconds with a diagnosis. Shared with model-e2e.yml. + run: bash .github/ci/kfd_health.sh /tmp/rocminfo.out - name: ROCm inventory includes ${{ matrix.arch }} run: cat /tmp/rocminfo.out | grep -E "Name:|Marketing Name:" | grep -F "${{ matrix.arch }}" diff --git a/.github/workflows/model-e2e.yml b/.github/workflows/model-e2e.yml new file mode 100644 index 000000000..5ce4efb80 --- /dev/null +++ b/.github/workflows/model-e2e.yml @@ -0,0 +1,307 @@ +name: Model e2e + +# Real model runs on lucebox3: build luce_server for one GPU, load a model, send +# a fixed prompt suite with greedy decoding (.github/ci/e2e/prompts.json) and +# compare with the last good run on main (the baseline). Qwen3.8-27B runs on the +# R9700, DeepSeek V4 Flash on the Strix Halo iGPU. Per-model files, flags and +# GPUs live in .github/ci/e2e/select_models.py. +# +# What fails the job: a crash, hang, GPU error in the kernel log, a server that will +# not load or stop, a failed request, or two or more checks that passed on the +# baseline and now fail. Changed output text, a single regressed check, slower +# decode or slower loading only warn: kernel changes legitimately flip near-tie +# tokens. See run_model_e2e.py for the details. +# +# When it runs: +# - only when a maintainer adds the `e2e` label to a PR. It runs the models the +# PR touches (select_models.py), or every model when it touches neither DS4, +# Qwen nor shared server code. To test new pushes, remove the label and add +# it again; +# - nightly on main, which refreshes the baselines, and on manual dispatch +# (refreshes them when `update_baseline` is ticked on main). +# +# Baselines: baseline runs on main upload their result as the artifact +# `model-e2e-baseline--`, and every job compares with the newest +# one (find_baseline.py). +# +# Bounded: 35 minutes per job. If someone is using lucebox3's GPUs by hand it +# waits up to 4 minutes, then skips with a warning rather than failing. A clean +# pass is remembered by tree hash, configuration and baseline, so re-running +# identical code against the same baseline is instant. +# +# Measured on lucebox5 (same hardware): a cold luce_server build takes ~100 s per +# architecture; the suite takes ~30 s for Qwen and ~80 s for DS4 including a warm +# model load. +# +# SECURITY: same model as ci.yml. Fork PRs run only after a maintainer approves +# the run, and only users with triage access can add the `e2e` label. Baselines +# are only taken from scheduled or dispatched runs on main, so PR code cannot +# replace them. The build directory and remembered passes persist on lucebox3 +# (~/.cache/lucebox-e2e), so approved PR code could tamper with them. Clear that +# directory if you suspect it. +# +# Models are read from vars.LUCEBOX_MODELS_DIR, default /opt/models. + +on: + pull_request: + branches: [main] + types: [labeled] + schedule: + - cron: '17 3 * * *' + workflow_dispatch: + inputs: + models: + description: Models to run + type: choice + options: [all, qwen, ds4] + default: all + update_baseline: + description: Upload passing results as the new baseline (main only) + type: boolean + default: false + +# Re-adding `e2e` replaces a PR's running e2e. The label is part of the group so +# that adding any other label (a run whose jobs all skip) cancels nothing. +concurrency: + group: model-e2e-${{ github.ref }}-${{ github.event.label.name || github.event_name }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + +permissions: + actions: read # find and download the baseline artifacts + contents: read + pull-requests: read + +jobs: + select: + name: Select models + if: github.event_name != 'pull_request' || github.event.label.name == 'e2e' + runs-on: ubuntu-latest + outputs: + matrix: ${{ steps.pick.outputs.matrix }} + any: ${{ steps.pick.outputs.any }} + steps: + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + with: + sparse-checkout: .github/ci/e2e + + - name: Pick models + id: pick + env: + GH_TOKEN: ${{ github.token }} + EVENT: ${{ github.event_name }} + PR: ${{ github.event.pull_request.number }} + DISPATCH_MODELS: ${{ inputs.models }} + run: | + set -euo pipefail + select=.github/ci/e2e/select_models.py + case "$EVENT" in + pull_request) + matrix=$(gh api "repos/$GITHUB_REPOSITORY/pulls/$PR/files" --paginate --jq '.[].filename' \ + | python3 "$select" --mode paths --fallback-all) + ;; + workflow_dispatch) + matrix=$(python3 "$select" --mode "$DISPATCH_MODELS") + ;; + *) matrix=$(python3 "$select" --mode all) ;; + esac + matrix=$(python3 .github/ci/e2e/find_baseline.py --repo "$GITHUB_REPOSITORY" <<<"$matrix") + echo "Selected: $matrix" + echo "matrix=$matrix" >> "$GITHUB_OUTPUT" + echo "any=$(jq '.include | length > 0' <<<"$matrix")" >> "$GITHUB_OUTPUT" + + e2e: + name: e2e ${{ matrix.model }} (${{ matrix.device_name }}) + needs: select + if: needs.select.outputs.any == 'true' + strategy: + fail-fast: false + matrix: ${{ fromJSON(needs.select.outputs.matrix) }} + # The runner runs one job at a time, so e2e jobs never share lucebox3's GPUs + # with each other or with gpu-tests-amd. + runs-on: [self-hosted, lucebox3] + timeout-minutes: 35 + env: + MODEL: ${{ matrix.model }} + DEVICE: ${{ matrix.device }} + ARCH: ${{ matrix.arch }} + HIP_VISIBLE_DEVICES: ${{ matrix.hip_index }} + SERVER_ARGS: ${{ matrix.server_args }} + OUT_DIR: ${{ github.workspace }}/e2e-out + UPDATE_BASELINE: ${{ github.ref == 'refs/heads/main' && (github.event_name == 'schedule' || inputs.update_baseline == true) }} + steps: + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + + - name: KFD health + run: | + set -euo pipefail + bash .github/ci/kfd_health.sh "$RUNNER_TEMP/rocminfo.out" + grep -qw "$ARCH" "$RUNNER_TEMP/rocminfo.out" || { echo "::error::no $ARCH GPU"; exit 1; } + + - name: Download the baseline + if: matrix.baseline_run != '' + uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 + with: + name: model-e2e-baseline-${{ matrix.model }}-${{ matrix.device }} + run-id: ${{ matrix.baseline_run }} + github-token: ${{ github.token }} + path: ${{ runner.temp }}/baseline + + - name: Check the models and earlier passes + id: cfg + env: + MODEL_DIR: ${{ vars.LUCEBOX_MODELS_DIR || '/opt/models' }} + TARGET_FILE: ${{ matrix.target }} + DRAFT_FILE: ${{ matrix.draft }} + run: | + set -euo pipefail + TARGET=$MODEL_DIR/$TARGET_FILE + DRAFT=${DRAFT_FILE:+$MODEL_DIR/$DRAFT_FILE} + BASELINE=$RUNNER_TEMP/baseline/baseline.json + state=$HOME/.cache/lucebox-e2e + mkdir -p "$state/passed" + { + echo "TARGET=$TARGET" + echo "DRAFT=$DRAFT" + echo "BASELINE=$BASELINE" + echo "BUILD_DIR=$state/build-$ARCH" + } >> "$GITHUB_ENV" + + # The draft is part of the tested configuration, not optional: without + # it the job would test target-only decoding and still look green. + missing="" + for f in "$TARGET" ${DRAFT:+"$DRAFT"}; do + if [ ! -f "$f" ]; then missing="$missing $f"; fi + done + if [ -n "$missing" ]; then + msg="Model weights not found on $(hostname):$missing" + echo "## Model e2e ($MODEL) skipped: $msg" >> "$GITHUB_STEP_SUMMARY" + if [ "$GITHUB_EVENT_NAME" = pull_request ]; then + echo "::warning title=Model e2e skipped::$msg" + echo "run=false" >> "$GITHUB_OUTPUT" + exit 0 + fi + echo "::error title=Model e2e unavailable::$msg" + exit 1 + fi + + # A result is remembered for everything that decides it: the tree, the + # model, device and flags, the model files, ROCm, and the baseline it + # was judged against (a new baseline can turn a pass into a failure). + key=$( + { + git rev-parse 'HEAD^{tree}' + printf '%s\n' "$MODEL" "$DEVICE" "$TARGET" "$DRAFT" "$SERVER_ARGS" + stat -c '%n %s %Y' "$TARGET" ${DRAFT:+"$DRAFT"} + cat /opt/rocm/.info/version 2>/dev/null || true + if [ -f "$BASELINE" ]; then sha256sum < "$BASELINE"; else echo "no baseline"; fi + } | sha256sum | cut -c1-32 + ) + echo "PASS_MARKER=$state/passed/$key" >> "$GITHUB_ENV" + if [ "$GITHUB_EVENT_NAME" = pull_request ] && [ -f "$state/passed/$key" ]; then + { + echo "This exact tree, configuration and baseline already passed cleanly; skipping." + echo "The report of that run:" + echo + cat "$state/passed/$key" + } >> "$GITHUB_STEP_SUMMARY" + echo "Already passed (key $key); skipping." + echo "run=false" >> "$GITHUB_OUTPUT" + exit 0 + fi + echo "run=true" >> "$GITHUB_OUTPUT" + + - name: Wait for the GPUs to be free + id: gpu + if: steps.cfg.outputs.run == 'true' + timeout-minutes: 6 + run: | + set -euo pipefail + bash .github/ci/e2e/gpu_wait.sh 240 | tee "$RUNNER_TEMP/gpu_wait.out" + grep '^state=' "$RUNNER_TEMP/gpu_wait.out" | tail -1 >> "$GITHUB_OUTPUT" + + - name: Build luce_server (${{ matrix.arch }}) + if: steps.gpu.outputs.state == 'free' + timeout-minutes: 12 + run: | + set -euo pipefail + # The build directory persists between runs so only changed files + # rebuild. Start over when it belongs to a different checkout path. + if [ -f "$BUILD_DIR/CMakeCache.txt" ] && + ! grep -qxF "CMAKE_HOME_DIRECTORY:INTERNAL=$PWD/server" "$BUILD_DIR/CMakeCache.txt"; then + rm -rf "$BUILD_DIR" + fi + launcher=() + if command -v ccache >/dev/null; then + launcher=(-DCMAKE_C_COMPILER_LAUNCHER=ccache -DCMAKE_CXX_COMPILER_LAUNCHER=ccache + -DCMAKE_HIP_COMPILER_LAUNCHER=ccache) + fi + cmake -S server -B "$BUILD_DIR" \ + -DLUCE_GPU_BACKEND=hip \ + -DLUCE_HIP_ARCHITECTURES="$ARCH" \ + -DGGML_HIP_GRAPHS=ON \ + -DCMAKE_BUILD_TYPE=Release \ + "${launcher[@]}" + cmake --build "$BUILD_DIR" --target luce_server --parallel "$(nproc)" + + - name: Run the prompt suite + id: run + if: steps.gpu.outputs.state == 'free' + timeout-minutes: 15 + run: | + set -euo pipefail + extra=() + if [ -n "$DRAFT" ]; then extra+=(--draft "$DRAFT"); fi + if [ "$UPDATE_BASELINE" = true ]; then extra+=(--write-baseline "$OUT_DIR/baseline.json"); fi + # DS4 loads in ~30 s from the page cache but ~185 s from disk. + python3 .github/ci/e2e/run_model_e2e.py \ + --model "$MODEL" --device "$DEVICE" \ + --server "$BUILD_DIR/luce_server" \ + --target "$TARGET" \ + --server-args "$SERVER_ARGS" \ + --port "$((18200 + HIP_VISIBLE_DEVICES))" \ + --baseline "$BASELINE" "${extra[@]}" \ + --load-timeout 480 --request-timeout 120 --budget 300 \ + --out-dir "$OUT_DIR" + + # Only a clean pass: a skipped re-run shows the old report but raises no + # warnings, so a warning must be seen again on every run. + - name: Remember the pass + if: steps.run.outcome == 'success' + run: | + verdict=$(python3 -c 'import json, sys; print(json.load(open(sys.argv[1]))["verdict"])' \ + "$OUT_DIR/result.json") + if [ "$verdict" = pass ]; then cp "$OUT_DIR/report.md" "$PASS_MARKER"; fi + + # The run writes baseline.json only when it did not fail. + - name: Upload the new baseline + if: steps.run.outcome == 'success' && env.UPDATE_BASELINE == 'true' + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: model-e2e-baseline-${{ matrix.model }}-${{ matrix.device }} + path: e2e-out/baseline.json + if-no-files-found: error + retention-days: 90 + + - name: Publish the report + if: always() + env: + GPU_STATE: ${{ steps.gpu.outputs.state }} + run: | + if [ -f "$OUT_DIR/report.md" ]; then + cat "$OUT_DIR/report.md" >> "$GITHUB_STEP_SUMMARY" + elif [ "$GPU_STATE" = busy ]; then + echo "## Model e2e ($MODEL) skipped: the GPUs were busy" >> "$GITHUB_STEP_SUMMARY" + fi + + - name: Stop leftover servers + if: always() + run: pkill -f "${BUILD_DIR:-/nonexistent}/luce_server" || true + + - name: Upload results + if: always() && steps.run.outcome != 'skipped' + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: model-e2e-${{ matrix.model }} + path: e2e-out/ + if-no-files-found: ignore + retention-days: 14 From 9cf78e160e1179dbcc834ba03b2334bf6f3b9581 Mon Sep 17 00:00:00 2001 From: Graffioh Date: Fri, 25 Sep 2026 06:46:44 +0000 Subject: [PATCH 3/3] ci(e2e): refresh the baselines on every merge to main Run the model e2e on each push to main instead of nightly, so a merged change that alters a model's output becomes the baseline for the next PRs. A push runs only the models whose code changed between their baseline's commit and the pushed commit (find_baseline.py --only-changed), so a merge that was skipped, replaced in the queue or failed is covered by the next one. Baselines now come from pushes to or dispatched runs on main. Co-Authored-By: Claude Opus 5.5 (1M context) --- .github/ci/e2e/README.md | 11 +++--- .github/ci/e2e/find_baseline.py | 65 +++++++++++++++++++++++++------- .github/ci/e2e/select_models.py | 6 ++- .github/ci/e2e/test_model_e2e.py | 33 ++++++++++++++-- .github/workflows/model-e2e.yml | 25 +++++++----- 5 files changed, 107 insertions(+), 33 deletions(-) diff --git a/.github/ci/e2e/README.md b/.github/ci/e2e/README.md index 0ca641227..995d516bb 100644 --- a/.github/ci/e2e/README.md +++ b/.github/ci/e2e/README.md @@ -25,11 +25,12 @@ any user can read. ## Baselines -The nightly run on main uploads each model's result as the artifact -`model-e2e-baseline--` (kept 90 days) unless it fails. Every job -compares with the newest one. To refresh a baseline by hand, e.g. after a change -that is meant to alter the output, run the workflow on main with -`update_baseline` ticked. Until the first baseline exists, jobs still fail on +Every merge to main runs the models whose code changed since their baseline +and uploads each result as the artifact `model-e2e-baseline--` +(kept 90 days) unless it fails. Every job compares with the newest one, so a +merged change that alters the output becomes the reference for the next PRs. To +refresh a baseline by hand, e.g. after a ROCm upgrade, run the workflow on main +with `update_baseline` ticked. Until the first baseline exists, jobs still fail on crashes, hangs and failed checks but cannot detect changed output. ## Benchmarking by hand on lucebox3 diff --git a/.github/ci/e2e/find_baseline.py b/.github/ci/e2e/find_baseline.py index 0d8a633b8..c20ce2f78 100644 --- a/.github/ci/e2e/find_baseline.py +++ b/.github/ci/e2e/find_baseline.py @@ -1,16 +1,22 @@ #!/usr/bin/env python3 """Find each model's baseline: the newest baseline artifact from a main run. -Baseline runs on main (nightly, or dispatched with update_baseline) upload +Baseline runs on main (every merge, or dispatched with update_baseline) upload their result as the artifact `model-e2e-baseline--`. This reads the select job's matrix on stdin and adds to each entry the `baseline_run` that holds its baseline, or "" when there is none yet. -Only artifacts from scheduled or dispatched model-e2e runs on this -repository's main count: pull request code runs in the same workflow and could -upload an artifact with the same name. +With --only-changed HEAD, it keeps only the models whose code changed between +their baseline's commit and HEAD (or that have no baseline): a merge that +cannot change a model's output does not need a new baseline. Measuring from the +baseline rather than the previous commit means a merge whose run was skipped +or failed is picked up by the next one. - python3 find_baseline.py --repo OWNER/REPO < matrix.json +Only artifacts from pushes to or dispatched runs on this repository's main +count: pull request code runs in the same workflow and could upload an artifact +with the same name. + + python3 find_baseline.py --repo OWNER/REPO [--only-changed SHA] < matrix.json """ from __future__ import annotations @@ -23,8 +29,12 @@ import urllib.request from collections.abc import Callable +from select_models import models_for + WORKFLOW = ".github/workflows/model-e2e.yml" -TRUSTED_EVENTS = ("schedule", "workflow_dispatch") +TRUSTED_EVENTS = ("push", "workflow_dispatch") +# The compare API lists at most this many files; a longer diff counts as all changed. +COMPARE_FILE_LIMIT = 300 Api = Callable[[str], dict] @@ -58,38 +68,65 @@ def trusted(run: dict, repo: str) -> bool: ) -def find_run(api: Api, repo: str, name: str) -> str: - """The newest trusted run that uploaded `name`, or "".""" +def find_run(api: Api, repo: str, name: str) -> dict: + """The newest trusted run that uploaded `name`, or {}.""" query = urllib.parse.urlencode({"name": name, "per_page": 50}) artifacts = api(f"repos/{repo}/actions/artifacts?{query}").get("artifacts", []) artifacts = [a for a in artifacts if a.get("name") == name and not a.get("expired")] artifacts.sort(key=lambda a: a.get("created_at") or "", reverse=True) for artifact in artifacts: run_id = (artifact.get("workflow_run") or {}).get("id") - if run_id and trusted(api(f"repos/{repo}/actions/runs/{run_id}"), repo): - return str(run_id) - return "" + if run_id: + run = api(f"repos/{repo}/actions/runs/{run_id}") + if trusted(run, repo): + return run + return {} def add_baselines(matrix: dict, api: Api, repo: str) -> dict: - found: dict[str, str] = {} + found: dict[str, dict] = {} for entry in matrix["include"]: name = artifact_name(entry) if name not in found: found[name] = find_run(api, repo, name) - entry["baseline_run"] = found[name] + entry["baseline_run"] = str(found[name].get("id") or "") + entry["baseline_commit"] = found[name].get("head_sha") or "" return matrix +def keep_changed(matrix: dict, api: Api, repo: str, head: str) -> dict: + """Drop the entries whose model code is unchanged since their baseline.""" + kept = [] + for entry in matrix["include"]: + base = entry["baseline_commit"] + if base: + files = api(f"repos/{repo}/compare/{base}...{head}").get("files", []) + paths = [f["filename"] for f in files] + if len(files) < COMPARE_FILE_LIMIT and entry["model"] not in models_for(paths): + continue + kept.append(entry) + return {"include": kept} + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) parser.add_argument("--repo", required=True, help="OWNER/REPO") + parser.add_argument( + "--only-changed", + metavar="SHA", + help="keep only the models whose code changed between their baseline and SHA", + ) args = parser.parse_args(argv) token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN") or "" - matrix = add_baselines(json.load(sys.stdin), github_api(token), args.repo) + api = github_api(token) + matrix = add_baselines(json.load(sys.stdin), api, args.repo) for entry in matrix["include"]: where = f"run {entry['baseline_run']}" if entry["baseline_run"] else "none yet" print(f"Baseline for {entry['model']}: {where}", file=sys.stderr) + if args.only_changed: + matrix = keep_changed(matrix, api, args.repo, args.only_changed) + kept = [e["model"] for e in matrix["include"]] + print(f"Changed since their baseline: {', '.join(kept) or 'none'}", file=sys.stderr) print(json.dumps(matrix, separators=(",", ":"))) return 0 diff --git a/.github/ci/e2e/select_models.py b/.github/ci/e2e/select_models.py index d86bc8653..ad3170bd8 100644 --- a/.github/ci/e2e/select_models.py +++ b/.github/ci/e2e/select_models.py @@ -3,7 +3,11 @@ Changed paths come on stdin, one per line. A path under a model's own sources selects that model; a path in shared server code selects every model; anything -else (docs, scripts, other models) selects nothing. +else (docs, scripts, other models) selects nothing. With --fallback-all, used +for a PR a maintainer labelled `e2e`, an empty selection runs every model: the +label is an explicit request, so a PR that only touches another model's code +still gets both runs. Other-model paths still keep a mixed PR (e.g. laguna + +qwen35) to the models it touches, and keep merges to main from re-running them. Each matrix entry carries everything its job needs: the GPU on lucebox3, the model files (in the models directory) and the server flags. diff --git a/.github/ci/e2e/test_model_e2e.py b/.github/ci/e2e/test_model_e2e.py index db8477965..efbb4cae8 100644 --- a/.github/ci/e2e/test_model_e2e.py +++ b/.github/ci/e2e/test_model_e2e.py @@ -13,7 +13,7 @@ import pytest import run_model_e2e -from find_baseline import add_baselines +from find_baseline import add_baselines, keep_changed from run_model_e2e import evaluate, kernel_lines_since, load_prompts, main, run_check from select_models import matrix, models_for @@ -432,19 +432,26 @@ def test_matrix_carries_each_models_gpu_and_files() -> None: REPO = "Luce-Org/lucebox" -def fake_api(artifacts: list[dict], runs: dict[int, dict]): +def fake_api( + artifacts: list[dict], runs: dict[int, dict], diffs: dict[str, list[str]] | None = None +): def get(path: str) -> dict: if "/actions/artifacts?" in path: name = path.split("name=")[1].split("&")[0] return {"artifacts": [a for a in artifacts if a["name"] == name]} - return runs[int(path.rsplit("/", 1)[1])] + if "/compare/" in path: + base = path.rsplit("/", 1)[1].split("...")[0] + return {"files": [{"filename": f} for f in (diffs or {})[base]]} + run_id = int(path.rsplit("/", 1)[1]) + return {**runs[run_id], "id": run_id} return get -def main_run(event: str = "schedule", **overrides: object) -> dict: +def main_run(event: str = "push", sha: str = "base", **overrides: object) -> dict: return { "event": event, + "head_sha": sha, "head_branch": "main", "head_repository": {"full_name": REPO}, "path": ".github/workflows/model-e2e.yml", @@ -478,6 +485,24 @@ def test_baseline_is_the_newest_from_a_trusted_main_run() -> None: assert {e["model"]: e["baseline_run"] for e in got["include"]} == {"ds4": "", "qwen": "2"} +@pytest.mark.parametrize( + ("changed", "kept"), + [ + (["server/src/qwen35/qwen35_target_graph.cpp"], ["ds4", "qwen"]), + (["docs/x.md"], ["ds4"]), + ([f"docs/{i}.md" for i in range(300)], ["ds4", "qwen"]), + ], +) +def test_merges_rerun_models_changed_since_their_baseline( + changed: list[str], kept: list[str] +) -> None: + # qwen has a baseline at commit "base"; ds4 has none yet, so it always runs. + artifacts = [artifact(1, "2026-09-01T03:00:00Z")] + api = fake_api(artifacts, {1: main_run()}, {"base": changed}) + got = keep_changed(add_baselines(matrix(["ds4", "qwen"]), api, REPO), api, REPO, "head") + assert [e["model"] for e in got["include"]] == kept + + def test_evaluate_warns_when_the_baseline_ran_on_another_rocm() -> None: base, now = result(prompt("a", "pass", "x")), result(prompt("a", "pass", "x")) base["rocm"], now["rocm"] = "7.1.0", "7.2.0" diff --git a/.github/workflows/model-e2e.yml b/.github/workflows/model-e2e.yml index 5ce4efb80..fc0865596 100644 --- a/.github/workflows/model-e2e.yml +++ b/.github/workflows/model-e2e.yml @@ -17,8 +17,10 @@ name: Model e2e # PR touches (select_models.py), or every model when it touches neither DS4, # Qwen nor shared server code. To test new pushes, remove the label and add # it again; -# - nightly on main, which refreshes the baselines, and on manual dispatch -# (refreshes them when `update_baseline` is ticked on main). +# - on every merge to main, for the models whose code changed since their +# baseline. A run that does not fail becomes the new baseline; +# - on manual dispatch (refreshes the baselines when `update_baseline` is +# ticked on main). # # Baselines: baseline runs on main upload their result as the artifact # `model-e2e-baseline--`, and every job compares with the newest @@ -35,7 +37,7 @@ name: Model e2e # # SECURITY: same model as ci.yml. Fork PRs run only after a maintainer approves # the run, and only users with triage access can add the `e2e` label. Baselines -# are only taken from scheduled or dispatched runs on main, so PR code cannot +# are only taken from pushes to or dispatched runs on main, so PR code cannot # replace them. The build directory and remembered passes persist on lucebox3 # (~/.cache/lucebox-e2e), so approved PR code could tamper with them. Clear that # directory if you suspect it. @@ -46,8 +48,8 @@ on: pull_request: branches: [main] types: [labeled] - schedule: - - cron: '17 3 * * *' + push: + branches: [main] workflow_dispatch: inputs: models: @@ -61,7 +63,9 @@ on: default: false # Re-adding `e2e` replaces a PR's running e2e. The label is part of the group so -# that adding any other label (a run whose jobs all skip) cancels nothing. +# that adding any other label (a run whose jobs all skip) cancels nothing. Runs +# on main never cancel a running one; a newer merge may replace a waiting one, +# which is fine because it runs everything changed since the baselines. concurrency: group: model-e2e-${{ github.ref }}-${{ github.event.label.name || github.event_name }} cancel-in-progress: ${{ github.event_name == 'pull_request' }} @@ -102,9 +106,12 @@ jobs: workflow_dispatch) matrix=$(python3 "$select" --mode "$DISPATCH_MODELS") ;; - *) matrix=$(python3 "$select" --mode all) ;; + push) matrix=$(python3 "$select" --mode all) ;; esac - matrix=$(python3 .github/ci/e2e/find_baseline.py --repo "$GITHUB_REPOSITORY" <<<"$matrix") + only_changed=() + if [ "$EVENT" = push ]; then only_changed=(--only-changed "$GITHUB_SHA"); fi + matrix=$(python3 .github/ci/e2e/find_baseline.py --repo "$GITHUB_REPOSITORY" \ + "${only_changed[@]}" <<<"$matrix") echo "Selected: $matrix" echo "matrix=$matrix" >> "$GITHUB_OUTPUT" echo "any=$(jq '.include | length > 0' <<<"$matrix")" >> "$GITHUB_OUTPUT" @@ -127,7 +134,7 @@ jobs: HIP_VISIBLE_DEVICES: ${{ matrix.hip_index }} SERVER_ARGS: ${{ matrix.server_args }} OUT_DIR: ${{ github.workspace }}/e2e-out - UPDATE_BASELINE: ${{ github.ref == 'refs/heads/main' && (github.event_name == 'schedule' || inputs.update_baseline == true) }} + UPDATE_BASELINE: ${{ github.ref == 'refs/heads/main' && (github.event_name == 'push' || inputs.update_baseline == true) }} steps: - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3