diff --git a/.github/ci/e2e/README.md b/.github/ci/e2e/README.md new file mode 100644 index 000000000..995d516bb --- /dev/null +++ b/.github/ci/e2e/README.md @@ -0,0 +1,47 @@ +# 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 + +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 + +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..c20ce2f78 --- /dev/null +++ b/.github/ci/e2e/find_baseline.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +"""Find each model's baseline: the newest baseline artifact from a main run. + +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. + +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. + +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 + +import argparse +import json +import os +import sys +import urllib.parse +import urllib.request +from collections.abc import Callable + +from select_models import models_for + +WORKFLOW = ".github/workflows/model-e2e.yml" +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] + + +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) -> 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: + 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, 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"] = 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 "" + 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 + + +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..ad3170bd8 --- /dev/null +++ b/.github/ci/e2e/select_models.py @@ -0,0 +1,130 @@ +#!/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. 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. + + 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..efbb4cae8 --- /dev/null +++ b/.github/ci/e2e/test_model_e2e.py @@ -0,0 +1,511 @@ +"""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, keep_changed +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], 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]} + 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 = "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", + **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"} + + +@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" + 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 876b9ec4d..284f6298e 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,40 +320,17 @@ 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 - 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..fc0865596 --- /dev/null +++ b/.github/workflows/model-e2e.yml @@ -0,0 +1,314 @@ +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; +# - 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 +# 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 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. +# +# Models are read from vars.LUCEBOX_MODELS_DIR, default /opt/models. + +on: + pull_request: + branches: [main] + types: [labeled] + push: + branches: [main] + 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. 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' }} + +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") + ;; + push) matrix=$(python3 "$select" --mode all) ;; + esac + 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" + + 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 == 'push' || 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 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: