chore: merge main into the page audit

scripts/generate_site.py

@@ -4329,6 +4329,8 @@ def generate_platform_page(
     no_list = not cov["total"] and registry.get(name, {}).get("native_list") == "none"
     if name in GUIDES:
         lines += _platform_guide(name, display, cov, registry, no_list)
+    # Before the reference: the links below it are read under that heading.
+    lines.extend(sitemeasurements.pack_tested_section(_PACK_ITEMS.get(name, [])))
 
     lines += [
         "## Reference: every file by system { #reference-every-file-by-system }",
@@ -9109,6 +9111,8 @@ _MEASUREMENT_ITEMS: dict[str, list[dict]] = {}
 # (manufacturer slug, console) -> the region outcomes of its systems
 # (region_check.py), shown under the console on its system page.
 _REGION_RESULTS: dict[tuple[str, str], list[dict]] = {}
+# Pack probes by platform key, for the Tested block of a platform page.
+_PACK_ITEMS: dict[str, list[dict]] = {}
 # The emulator pages that list each data directory, by ref.
 _DATA_READERS: dict[str, list[tuple[str, str]]] = {}
 
@@ -11896,6 +11900,8 @@ def _generate(args: argparse.Namespace) -> None:
         _record(path)
     _MEASUREMENT_ITEMS.clear()
     _MEASUREMENT_ITEMS.update(sitemeasurements.by_emulator(measured.index))
+    _PACK_ITEMS.clear()
+    _PACK_ITEMS.update(sitemeasurements.by_platform(measured.index))
     print(f"  {measured.index['records']} records, {measured.index['count']} tests")
     write_if_changed(str(docs / sitemeasurements.EXPLORER_PAGE),
                      sitemeasurements.generate_explorer_page(

scripts/measurement_records.py

@@ -84,13 +84,17 @@ def records_path(emulator: str) -> str:
 
 
 # The campaign's own files beside the records: append-only like them, one
-# line per finding or per pass, each line hashed to its id. No profile takes
-# these names (tests/test_campaign.py checks it).
+# line per finding, per pass or per pack probe, each line hashed to its id.
+# No profile takes these names (tests/test_campaign.py checks it).
 FINDINGS = "findings"
 CAMPAIGN_RUNS = "campaign-runs"
+PACK_PROBES = "pack-probes"
+# The subject of every core_probe record a pack probe writes.
+PACK_SUBJECT_PREFIX = "pack-"
 CAMPAIGN_FILES = {
     FINDINGS: ("found_at", "campaign-finding.schema.json"),
     CAMPAIGN_RUNS: ("started_at", "campaign-run.schema.json"),
+    PACK_PROBES: ("probed_at", "pack-probe.schema.json"),
 }
 
 

scripts/pack_probe.py

@@ -0,0 +1,712 @@
+#!/usr/bin/env python3
+"""Test a platform's pack the way a user installs it.
+
+For one platform, the probe lays the platform's BIOS folder out in a scratch
+root, in one or both of the ways a user gets it:
+
+  pack       the pack generate_pack.py builds for the platform, extracted
+  installer  install.py --dest on the scratch root, from that pack and the
+             published manifest (install/<platform>.json), with no network
+
+Then for each system a test game covers (tests/campaign/content.yml and
+tests/campaign/pack_content.yml), it resolves the libretro cores the platform
+runs for that system (platform config, target file, standalone and dual-mode
+lists), and runs the game through core_probe.py with the system directory
+taken from that layout.
+
+core_probe copies what it stages, so the probe stages the files the core can
+reach rather than the whole folder: the files the core's profile and the
+platform's list declare, every file beside them in their directory, and for
+a core that finds its BIOS by content, every root file of a size it accepts.
+A file the core then looks for that the layout holds is staged and the run
+repeated, until the core asks for nothing more the layout has.
+
+Each run is a core_probe record in measurements/<emulator>.jsonl and one line
+of measurements/pack-probes.jsonl naming the platform, the pack or manifest,
+the verdict, every file the core opened mapped to the pack entry that put it
+there, and the findings: a file the core looked for that the layout lacks,
+or holds under another path.
+
+Each core_probe run holds the host's probe lock (~/.local/state/retrobios/
+probe.lock, RETROBIOS_PROBE_LOCK) and releases it before the next: other
+probes run between ours. core_probe is told it is held
+(RETROBIOS_PROBE_LOCK_HELD). Building and extracting the pack take no lock.
+
+Usage:
+    python scripts/pack_probe.py run --platform retroarch [--system sony-playstation]
+    python scripts/pack_probe.py plan --platform batocera
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import datetime as dt
+import fcntl
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import time
+import urllib.request
+import zipfile
+from collections.abc import Iterable, Iterator, Sequence
+from dataclasses import dataclass, field
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+import autorun
+import campaign
+import measurement_records as mr
+from common import (
+    _norm_system_id,
+    load_emulator_profiles,
+    load_platform_config,
+    load_target_config,
+    platform_arg,
+    require_platform,
+    resolve_platform_cores,
+    run_cli,
+    runs_standalone,
+)
+from validation import filter_files_by_mode
+from ziptools import safe_extract_zip
+
+ROOT = Path(__file__).resolve().parent.parent
+MAIN = autorun.main_checkout()
+PACK_CONTENT_FILE = ROOT / "tests" / "campaign" / "pack_content.yml"
+CONTENT_DIR = MAIN / ".cache" / "test-content"
+CORE_CACHE = MAIN / ".cache" / "pack-probe" / "cores"
+WORK_DIR = ROOT / "tmp" / "pack_probe"
+BUILDBOT = campaign.BUILDBOT
+PACK_PROBES = mr.PACK_PROBES
+INSTALLS = ("pack", "installer")
+VERDICTS = ("boots", "bios_error", "crash", "hang", "unknown")
+MAX_ROUNDS = 6
+PROBE_TIMEOUT = 900
+IDLE_WAIT = 1800
+PROBE_LOCK_ENV = "RETROBIOS_PROBE_LOCK"
+PROBE_LOCK_HELD_ENV = "RETROBIOS_PROBE_LOCK_HELD"
+PROBE_LOCK = Path(os.environ.get(PROBE_LOCK_ENV)
+                  or Path.home() / ".local" / "state" / "retrobios" / "probe.lock")
+EXIT_BUSY = 3
+
+# The hardware target each platform is probed for: the one whose libretro
+# cores are the buildbot's x86_64 Linux builds, or the platform's only one.
+TARGETS = {"retroarch": "linux-x86_64", "batocera": "x86_64", "retrobat": "windows"}
+
+
+class ProbeFailure(RuntimeError):
+    """The probe cannot go on: no pack, no installer layout, a busy host."""
+
+
+# --------------------------------------------------------------------------
+# Layouts: the platform's BIOS folder as a user ends up with it
+# --------------------------------------------------------------------------
+
+@dataclass
+class Layout:
+    """A BIOS folder: every file by its path under the folder, and the pack
+    or manifest entry that put it there (None for a file nothing declares)."""
+
+    install: str
+    root: Path
+    files: dict[str, str | None]
+    source: dict
+    sizes: dict[str, int] = field(default_factory=dict)
+
+    def size(self, rel: str) -> int:
+        if rel not in self.sizes:
+            self.sizes[rel] = (self.root / rel).stat().st_size
+        return self.sizes[rel]
+
+
+def walk(root: Path) -> list[str]:
+    return sorted(p.relative_to(root).as_posix() for p in root.rglob("*") if p.is_file())
+
+
+def file_sha(path: Path, algorithm: str = "sha1") -> str:
+    digest = hashlib.new(algorithm)
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1 << 20), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def pack_layout(pack: Path, root: Path, identity: dict) -> Layout:
+    """The pack extracted as a user does, at the root of the BIOS folder.
+    Each file's entry is the archive member it came from."""
+    if root.exists():
+        shutil.rmtree(root)
+    safe_extract_zip(str(pack), str(root))
+    with zipfile.ZipFile(pack) as handle:
+        members = {info.filename for info in handle.infolist() if not info.is_dir()}
+    files = {rel: rel if rel in members else None for rel in walk(root)}
+    return Layout("pack", root, files, {"pack": identity, "manifest": None})
+
+
+def pack_identity(pack: Path) -> dict:
+    with zipfile.ZipFile(pack) as handle:
+        members = sum(1 for info in handle.infolist() if not info.is_dir())
+    return {"file": pack.name, "sha256": file_sha(pack, "sha256"), "size": pack.stat().st_size,
+            "members": members}
+
+
+def manifest_entries(manifest: dict) -> dict[str, str]:
+    """Destination -> the manifest entry that names it, as the installer
+    places it under --dest: a file by its dest, a data directory member
+    under its directory's dest."""
+    out: dict[str, str] = {}
+    for item in manifest.get("files") or ():
+        out[item["dest"]] = f"files:{item['dest']}"
+    for item in manifest.get("case_variants") or ():
+        out.setdefault(item["dest"], f"case_variants:{item['dest']}")
+    for directory in manifest.get("data_directories") or ():
+        base = str(directory.get("dest") or "").strip("/")
+        for member in directory.get("files") or {}:
+            dest = f"{base}/{member}" if base else member
+            out.setdefault(dest, f"data_directories:{directory['ref']}:{member}")
+    return out
+
+
+def installer_layout(platform: str, pack: Path, identity: dict, manifest_path: Path,
+                     root: Path, narrow: Sequence[str] = ()) -> Layout:
+    """install.py --dest on a scratch root, from the pack, offline. The
+    manifest is the published one, the file a user's installer reads.
+    ``narrow`` passes install.py options such as --system, for the tests."""
+    system_dir, save_dir = root / "system", root / "saves"
+    if root.exists():
+        shutil.rmtree(root)
+    argv = [sys.executable, str(ROOT / "install.py"), "--platform", platform, "--dest", str(system_dir),
+            "--save-dir", str(save_dir), "--from-pack", str(pack), "--manifest", str(manifest_path),
+            "--offline", "--yes", "--quiet", *narrow]
+    proc = subprocess.run(argv, capture_output=True, text=True, check=False, cwd=ROOT)
+    if proc.returncode != 0 or not system_dir.is_dir():
+        raise ProbeFailure(f"install.py exited {proc.returncode}: {(proc.stderr or proc.stdout)[-1500:]}")
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    entries = manifest_entries(manifest)
+    files = {rel: entries.get(rel) for rel in walk(system_dir)}
+    source = {
+        "pack": identity,
+        "manifest": {"path": manifest_path.relative_to(ROOT).as_posix() if manifest_path.is_relative_to(ROOT)
+                     else manifest_path.name,
+                     "sha256": file_sha(manifest_path, "sha256"),
+                     "generated": manifest.get("generated"), "pack_files": manifest.get("pack_files")},
+        "save_dir_files": len(walk(save_dir)) if save_dir.is_dir() else 0,
+        "installer_summary": " ".join(proc.stdout.split())[-400:],
+    }
+    return Layout("installer", system_dir, files, source)
+
+
+def build_pack(platform: str, out_dir: Path, narrow: Sequence[str] = ()) -> Path:
+    """The platform's pack, built from this checkout as a release builds it.
+    ``narrow`` passes generate_pack.py options such as --system, for the tests."""
+    out_dir.mkdir(parents=True, exist_ok=True)
+    for old in out_dir.glob("*_BIOS_Pack*.zip"):
+        old.unlink()
+    argv = [sys.executable, str(ROOT / "scripts" / "generate_pack.py"), "--platform", platform,
+            "--offline", "--output-dir", str(out_dir), *narrow]
+    proc = subprocess.run(argv, capture_output=True, text=True, check=False, cwd=ROOT)
+    packs = sorted(p for p in out_dir.glob("*_BIOS_Pack*.zip") if ".part" not in p.name)
+    if proc.returncode != 0 or len(packs) != 1:
+        raise ProbeFailure(f"generate_pack.py exited {proc.returncode} with {len(packs)} packs: "
+                           f"{proc.stderr[-1500:]}")
+    return packs[0]
+
+
+# --------------------------------------------------------------------------
+# Which cores a platform runs for a system
+# --------------------------------------------------------------------------
+
+@dataclass(frozen=True)
+class CoreChoice:
+    emulator: str
+    so_name: str
+
+
+def platform_system(config: dict, system: str) -> str | None:
+    """The platform's own id of a system, compared on the normalised id."""
+    wanted = _norm_system_id(system)
+    matches = sorted(s for s in config.get("systems") or {} if _norm_system_id(s) == wanted)
+    return matches[0] if matches else None
+
+
+def libretro_build_runs(name: str, profile: dict, config: dict) -> bool:
+    """The platform launches this profile's libretro build: it has one, and
+    the platform does not run the emulator standalone only."""
+    if "libretro" not in str(profile.get("type", "")) or profile.get("type") == "alias":
+        return False
+    standalone = {str(c) for c in config.get("standalone_cores") or ()}
+    if not runs_standalone(name, profile, standalone):
+        return True
+    dual = {str(c) for c in config.get("dual_mode_cores") or ()}
+    return bool(dual & ({name} | {str(c) for c in profile.get("cores") or ()}))
+
+
+def so_names(name: str, profile: dict, buildbot: set[str], target: set[str] | None) -> list[str]:
+    """The buildbot builds of a profile the platform runs. Every one the
+    target names, else the build that carries the profile's key, else the
+    first of its cores: that is on the buildbot."""
+    names = [str(c) for c in profile.get("cores") or () if str(c) in buildbot]
+    if target is not None:
+        named = [n for n in names if n in target]
+        if named:
+            return sorted(named)
+    if name in buildbot:
+        return [name]
+    return names[:1]
+
+
+def choose_cores(config: dict, profiles: dict[str, dict], system: str, buildbot: set[str],
+                 target: set[str] | None) -> list[CoreChoice]:
+    wanted = _norm_system_id(system)
+    out = []
+    for name in sorted(resolve_platform_cores(config, profiles, target)):
+        profile = profiles[name]
+        if wanted not in {_norm_system_id(s) for s in profile.get("systems") or ()}:
+            continue
+        if not libretro_build_runs(name, profile, config):
+            continue
+        out.extend(CoreChoice(name, so) for so in so_names(name, profile, buildbot, target))
+    return out
+
+
+# --------------------------------------------------------------------------
+# Staging: what the core can reach, grown to a fixed point
+# --------------------------------------------------------------------------
+
+def entry_paths(entry: dict) -> list[str]:
+    """The paths a profile entry names under the system directory: its
+    path or name, and its aliases beside it."""
+    main = campaign.entry_dest(entry)
+    folder = main.rsplit("/", 1)[0] + "/" if "/" in main else ""
+    return [main] + [f"{folder}{alias}" for alias in entry.get("aliases") or ()]
+
+
+def libretro_entries(profile: dict) -> list[dict]:
+    return filter_files_by_mode(profile.get("files") or [], standalone=False)
+
+
+def declared_sizes(profile: dict) -> set[int]:
+    sizes: set[int] = set()
+    for entry in libretro_entries(profile):
+        value = entry.get("size")
+        sizes.update(v for v in (value if isinstance(value, list) else [value]) if isinstance(v, int))
+    return sizes
+
+
+def with_folders(files: Iterable[str], staged: set[str]) -> set[str]:
+    """Every file of every folder a staged file sits in, below the root: a
+    core that lists its folder sees what the platform's folder holds."""
+    folders = {rel.rsplit("/", 1)[0] + "/" for rel in staged if "/" in rel}
+    return set(staged) | {rel for rel in files if any(rel.startswith(f) for f in folders)}
+
+
+def initial_stage(layout: Layout, profile: dict, platform_dests: Iterable[str]) -> set[str]:
+    files = set(layout.files)
+    staged = {p for entry in libretro_entries(profile) for p in entry_paths(entry) if p in files}
+    staged |= {d for d in platform_dests if d in files}
+    if profile.get("found_by") == "content":
+        sizes = declared_sizes(profile)
+        staged |= {rel for rel in files if "/" not in rel and layout.size(rel) in sizes}
+    return with_folders(files, staged)
+
+
+def grow_stage(files: Iterable[str], staged: set[str], probed: Iterable[str]) -> set[str]:
+    """The files the layout holds that the core looked for, as a file or as
+    a folder, and their folders: empty once the staging is closed."""
+    files = set(files)
+    wanted = set()
+    for path in probed:
+        if path in files:
+            wanted.add(path)
+        else:
+            wanted |= {rel for rel in files if rel.startswith(path.rstrip("/") + "/")}
+    return with_folders(files, staged | wanted) - staged
+
+
+# --------------------------------------------------------------------------
+# Reading a run
+# --------------------------------------------------------------------------
+
+def run_verdict(record: dict | None) -> tuple[str, dict | None]:
+    """The verdict of a run record: variants[0] from record version 4,
+    unknown before the oracle or without a record."""
+    if not record or not record.get("variants"):
+        return "unknown", None
+    variant = record["variants"][0]
+    verdict = variant.get("verdict")
+    if verdict not in VERDICTS:
+        return "unknown", None
+    return verdict, variant.get("verdict_evidence")
+
+
+def first_repetition(record: dict) -> dict:
+    return record["variants"][0]["repetitions"][0]
+
+
+def declaring(profile: dict, path: str) -> list[str]:
+    return sorted({entry["name"] for entry in libretro_entries(profile) if path in entry_paths(entry)})
+
+
+def map_run(layout: Layout, staged: set[str], rep: dict, profile: dict) -> tuple[list[dict], list[dict]]:
+    """The files the core opened, each with its pack entry, and the findings."""
+    opened = []
+    for item in rep.get("files_opened") or ():
+        path = item["path"]
+        opened.append({"path": path, "entry": layout.files.get(path) if path in staged else None,
+                       "sha1": item.get("sha1_after"), "created": path not in staged})
+    by_name: dict[str, list[str]] = {}
+    for rel in layout.files:
+        by_name.setdefault(rel.rsplit("/", 1)[-1].casefold(), []).append(rel)
+    # An entry the core found under one of its names asks for nothing more:
+    # BlastEm tries tmss.md, then reads bios_MD.bin, both names of one entry.
+    read = {item["path"] for item in opened if not item["created"]}
+    satisfied = {entry["name"] for entry in libretro_entries(profile) if read & set(entry_paths(entry))}
+    findings = []
+    for path in rep.get("files_probed_absent") or ():
+        names = declaring(profile, path)
+        if path in layout.files:
+            findings.append({"kind": "not_staged", "path": path, "declared_by": names, "paths": [path]})
+            continue
+        if names and set(names) <= satisfied:
+            continue
+        elsewhere = sorted(p for p in by_name.get(path.rsplit("/", 1)[-1].casefold(), ()) if p != path)
+        if elsewhere:
+            kind = "elsewhere"
+        else:
+            kind = "missing_declared" if names else "missing_undeclared"
+        findings.append({"kind": kind, "path": path, "declared_by": names, "paths": elsewhere})
+    for item in opened:
+        if item["path"] in staged and item["entry"] is None:
+            findings.append({"kind": "unlisted", "path": item["path"], "declared_by": [], "paths": []})
+    return opened, findings
+
+
+# --------------------------------------------------------------------------
+# Cores and content
+# --------------------------------------------------------------------------
+
+def buildbot_listing() -> set[str]:
+    with urllib.request.urlopen(BUILDBOT, timeout=120) as response:
+        page = response.read().decode("utf-8", "replace")
+    return set(re.findall(r"([A-Za-z0-9_]+)_libretro\.so\.zip", page))
+
+
+def fetch_core(name: str, refresh: bool = False) -> dict:
+    """The buildbot nightly of a core, kept with where and when it came from."""
+    folder = CORE_CACHE / name
+    so, meta_path = folder / f"{name}_libretro.so", folder / "build.json"
+    if so.is_file() and meta_path.is_file() and not refresh:
+        meta = json.loads(meta_path.read_text(encoding="utf-8"))
+        if meta.get("sha256") == file_sha(so, "sha256"):
+            return {**meta, "path": str(so)}
+    url = f"{BUILDBOT}{name}_libretro.so.zip"
+    folder.mkdir(parents=True, exist_ok=True)
+    archive = folder / f"{name}.zip.part"
+    with urllib.request.urlopen(url, timeout=600) as response, archive.open("wb") as out:
+        shutil.copyfileobj(response, out)
+        modified = response.headers.get("Last-Modified")
+    campaign.safe_extract_so(archive, f"{name}_libretro.so", so)
+    archive.unlink()
+    meta = {"name": name, "url": url, "last_modified": modified, "sha256": file_sha(so, "sha256"),
+            "fetched_at": now_iso()}
+    meta_path.write_text(json.dumps(meta, indent=1) + "\n", encoding="utf-8")
+    return {**meta, "path": str(so)}
+
+
+def load_contents(files: Sequence[Path]) -> dict[str, list[campaign.Content]]:
+    out: dict[str, list[campaign.Content]] = {}
+    for path in files:
+        if path.is_file():
+            for system, items in campaign.load_content(path).items():
+                out.setdefault(system, []).extend(items)
+    return out
+
+
+def content_folder(content: campaign.Content, fetch: bool) -> Path:
+    """The game's folder, fetched through romfetch when absent and asked to."""
+    folder = CONTENT_DIR / content.romfetch_system / content.name
+    if not (folder / "identity.json").is_file() and fetch:
+        romfetch = shutil.which("romfetch")
+        if not romfetch:
+            raise campaign.Unavailable("romfetch not found")
+        proc = subprocess.run([romfetch, "get", content.romfetch_system, content.name, "--json", "-q",
+                               "--content-dir", str(CONTENT_DIR)], capture_output=True, text=True,
+                              check=False, env=campaign.child_env(romfetch=True))
+        if proc.returncode != 0:
+            raise campaign.Unavailable(f"romfetch exit {proc.returncode} for {content.name!r}")
+    campaign.verify_identity(folder)
+    return folder
+
+
+# --------------------------------------------------------------------------
+# One probe: a core, a game, a layout
+# --------------------------------------------------------------------------
+
+def now_iso() -> str:
+    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()
+
+
+def wait_idle(limit: float = IDLE_WAIT, poll: float = 30.0) -> None:
+    """Wait for the load core_probe records under, a quarter of the CPUs.
+    Extracting a pack loads the host itself for a minute or two."""
+    threshold = (os.cpu_count() or 1) / 4
+    waited = 0.0
+    while os.getloadavg()[0] > threshold:
+        if waited >= limit:
+            raise ProbeFailure(f"busy host: load {os.getloadavg()[0]:.2f} above {threshold} "
+                               f"for {int(waited)} s")
+        time.sleep(poll)
+        waited += poll
+
+
+@contextlib.contextmanager
+def probe_lock(path: Path | None = None) -> Iterator[None]:
+    """The host's probe lock, for one run: one probe at a time on the host,
+    released between runs so a pass never holds it for its whole length."""
+    path = path or PROBE_LOCK
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a") as handle:
+        fcntl.flock(handle, fcntl.LOCK_EX)
+        try:
+            yield
+        finally:
+            fcntl.flock(handle, fcntl.LOCK_UN)
+
+
+def core_probe(argv: list[str], staged: set[str], layout: Layout, work: Path) -> tuple[int, dict | None, str]:
+    """core_probe.py run under the probe lock, the system files passed in an
+    argument file."""
+    work.mkdir(parents=True, exist_ok=True)
+    argfile = work / "system-files.args"
+    argfile.write_text("".join(f"--system-file\n{rel}={layout.root / rel}\n" for rel in sorted(staged)),
+                       encoding="utf-8")
+    command = [sys.executable, str(ROOT / "scripts" / "core_probe.py"), *argv, f"@{argfile}"]
+    with probe_lock():
+        if "--record" in argv:
+            wait_idle()
+        # core_probe takes the same lock unless told its caller holds it.
+        proc = subprocess.run(command, capture_output=True, text=True, check=False, cwd=ROOT,
+                              timeout=PROBE_TIMEOUT * 2 + 120,
+                              env={**os.environ, PROBE_LOCK_HELD_ENV: "1", PROBE_LOCK_ENV: str(PROBE_LOCK)})
+    record = None
+    if proc.returncode == 0:
+        try:
+            record = json.loads(proc.stdout)
+        except ValueError:
+            record = None
+    return proc.returncode, record, proc.stderr.strip()[-1500:]
+
+
+def probe_argv(core: dict, choice: CoreChoice, content: campaign.Content, load: Path,
+               subject: str, record: bool, supersedes: str | None = None) -> list[str]:
+    argv = ["run", "--core", core["path"], "--content", str(load), "--content-label", content.name,
+            "--frames", str(content.frames), "--frame-hash-every", str(content.frame_hash_every),
+            "--require-sandbox", "--timeout", str(PROBE_TIMEOUT),
+            "--work-dir", str(ROOT / "tmp" / "core_probe"), "--core-artifact", core["url"]]
+    if record:
+        argv += ["--record", str(ROOT), "--emulator", choice.emulator, "--subject", subject]
+        if supersedes:
+            argv += ["--supersedes", supersedes]
+    return argv
+
+
+def subject_for(platform: str, install: str, system: str, so: str, content: str) -> str:
+    subject = mr.slugify(f"{platform} {install} {system} {so} {content}", 85)
+    return f"{mr.PACK_SUBJECT_PREFIX}{subject}"
+
+
+def probe_one(layout: Layout, platform: str, config: dict, profile: dict, choice: CoreChoice,
+              core: dict, system: str, content: campaign.Content, load: Path, work: Path) -> dict:
+    """Run until the staging is closed. The first run is recorded, and when
+    the staging grew, the closed one is recorded too and supersedes it."""
+    native = platform_system(config, system)
+    dests = [f.get("destination") or f.get("name")
+             for f in ((config.get("systems") or {}).get(native) or {}).get("files") or ()]
+    staged = initial_stage(layout, profile, dests)
+    subject = subject_for(platform, layout.install, system, choice.so_name, content.name)
+    code, record, error = core_probe(probe_argv(core, choice, content, load, subject, True),
+                                     staged, layout, work)
+    if code == EXIT_BUSY:
+        raise ProbeFailure(f"core_probe refused on a busy host: {error}")
+    rounds, first_id, records = 1, None, []
+    if record:
+        first_id = record["id"]
+        records.append(first_id)
+        added = grow_stage(layout.files, staged, first_repetition(record).get("files_probed_absent") or ())
+        while added and rounds < MAX_ROUNDS:
+            staged |= added
+            rounds += 1
+            code, record, error = core_probe(probe_argv(core, choice, content, load, subject, False),
+                                             staged, layout, work)
+            if not record:
+                break
+            added = grow_stage(layout.files, staged, first_repetition(record).get("files_probed_absent") or ())
+        if rounds > 1 and record:
+            code, record, error = core_probe(probe_argv(core, choice, content, load, subject, True, first_id),
+                                             staged, layout, work)
+            if code == EXIT_BUSY:
+                raise ProbeFailure(f"core_probe refused on a busy host: {error}")
+            if record:
+                records.append(record["id"])
+    verdict, evidence = run_verdict(record)
+    opened, findings = map_run(layout, staged, first_repetition(record), profile) if record else ([], [])
+    return {
+        "schema_version": 1,
+        "probed_at": now_iso(),
+        "platform": platform,
+        "platform_version": str(config.get("version") or ""),
+        "target": TARGETS.get(platform),
+        "install": layout.install,
+        "source": {**layout.source, "repo_commit": git_head()},
+        "system": system,
+        "platform_system": native,
+        "emulator": choice.emulator,
+        "core": {"file": f"{choice.so_name}_libretro.so", "sha256": core["sha256"], "artifact": core["url"],
+                 "last_modified": core.get("last_modified")},
+        "content": {"label": content.name, "catalog": content.catalog, "region": content.region},
+        "records": records,
+        "rounds": rounds,
+        "staged": len(staged),
+        "verdict": verdict,
+        "verdict_evidence": evidence,
+        "files_opened": opened,
+        "findings": findings,
+        "error": None if record else (error or f"core_probe exited {code}"),
+    }
+
+
+def git_head() -> str | None:
+    proc = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, check=False,
+                          cwd=ROOT)
+    return proc.stdout.strip() or None
+
+
+def append_line(line: dict, root: Path = ROOT) -> dict:
+    sealed = mr.seal_line(line, mr.CAMPAIGN_FILES[PACK_PROBES][0])
+    path = root / mr.records_path(PACK_PROBES)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(mr.line_text(sealed) + "\n")
+    return sealed
+
+
+# --------------------------------------------------------------------------
+# Plans and the command line
+# --------------------------------------------------------------------------
+
+@dataclass
+class Unit:
+    system: str
+    content: campaign.Content
+    cores: list[CoreChoice]
+
+
+def plan(platform: str, systems: set[str] | None, cores: set[str] | None,
+         buildbot: set[str]) -> tuple[dict, dict, list[Unit]]:
+    config = load_platform_config(platform, str(ROOT / "platforms"))
+    profiles = load_emulator_profiles(str(ROOT / "emulators"))
+    target = load_target_config(platform, TARGETS[platform], str(ROOT / "platforms"))
+    units = []
+    for system, contents in sorted(load_contents([campaign.CONTENT_FILE, PACK_CONTENT_FILE]).items()):
+        if systems and system not in systems:
+            continue
+        if not platform_system(config, system):
+            continue
+        choices = [c for c in choose_cores(config, profiles, system, buildbot, target)
+                   if not cores or c.so_name in cores or c.emulator in cores]
+        units.extend(Unit(system, content, choices) for content in contents)
+    return config, profiles, units
+
+
+def run(args: argparse.Namespace) -> int:
+    buildbot = buildbot_listing()
+    config, profiles, units = plan(args.platform, args.systems, args.cores, buildbot)
+    if args.mode == "plan":
+        for unit in units:
+            print(f"{unit.system}: {unit.content.name}: "
+                  f"{', '.join(f'{c.so_name} ({c.emulator})' for c in unit.cores) or 'no core'}")
+        return 0
+    if not units:
+        print(f"ERROR: no test game for any system of {args.platform}", file=sys.stderr)
+        return 2
+    work = Path(args.work_dir) / args.platform
+    pack = Path(args.pack) if args.pack else build_pack(args.platform, work / "dist")
+    identity = pack_identity(pack)
+    status = 0
+    for install in args.install:
+        if install == "pack":
+            layout = pack_layout(pack, work / "pack", identity)
+        else:
+            layout = installer_layout(args.platform, pack, identity,
+                                      ROOT / "install" / f"{args.platform}.json", work / "installer")
+        try:
+            for unit in units:
+                try:
+                    load = campaign.load_file(content_folder(unit.content, args.fetch), unit.content.load)
+                except campaign.Unavailable as exc:
+                    print(f"SKIP {unit.system}: {exc}", file=sys.stderr)
+                    continue
+                for choice in unit.cores:
+                    core = fetch_core(choice.so_name, args.refresh_cores)
+                    line = probe_one(layout, args.platform, config, profiles[choice.emulator], choice,
+                                     core, unit.system, unit.content, load, work / "runs")
+                    sealed = append_line(line)
+                    print(f"{args.platform} {install} {unit.system} {choice.so_name}: {line['verdict']}, "
+                          f"{len(line['files_opened'])} opened, {len(line['findings'])} findings, "
+                          f"line {sealed['id']}")
+                    if line["error"]:
+                        status = 1
+        finally:
+            if not args.keep:
+                shutil.rmtree(layout.root if install == "pack" else layout.root.parent, ignore_errors=True)
+    return status
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(allow_abbrev=False, description=__doc__,
+                                     formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("mode", choices=("run", "plan"),
+                        help="run: lay the folder out and start the games. plan: print the "
+                             "systems, games and cores, run nothing")
+    parser.add_argument("--platform", "-p", required=True, type=platform_arg,
+                        help=f"the platform whose pack is tested: {', '.join(sorted(TARGETS))}")
+    parser.add_argument("--system", dest="systems", action="append", default=[],
+                        help="only this system (repeat)")
+    parser.add_argument("--core", dest="cores", action="append", default=[],
+                        help="only this core, by buildbot name or profile (repeat)")
+    parser.add_argument("--install", action="append", choices=INSTALLS, default=[],
+                        help="pack, installer or both (default both)")
+    parser.add_argument("--pack", help="a pack already built for the platform, instead of building one")
+    parser.add_argument("--fetch", action="store_true", help="fetch a missing game through romfetch")
+    parser.add_argument("--refresh-cores", action="store_true", help="download the nightlies again")
+    parser.add_argument("--keep", action="store_true", help="keep the extracted layouts")
+    parser.add_argument("--work-dir", default=str(WORK_DIR),
+                        help="where the pack is built and laid out (default tmp/pack_probe)")
+    return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    require_platform(args.platform, str(ROOT / "platforms"))
+    if args.platform not in TARGETS:
+        print(f"ERROR: pack_probe tests {', '.join(sorted(TARGETS))}, not {args.platform}",
+              file=sys.stderr)
+        return 2
+    args.systems = set(args.systems) or None
+    args.cores = set(args.cores) or None
+    args.install = args.install or list(INSTALLS)
+    try:
+        return run(args)
+    except ProbeFailure as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return EXIT_BUSY if "busy host" in str(exc) else 1
+
+
+if __name__ == "__main__":
+    run_cli(main)

scripts/sitemarkup.py

@@ -75,11 +75,25 @@ def inline_html(text: str) -> str:
             continue
         part = html.escape(part, quote=False)
         part = re.sub(r"\*\*([^*]+)\*\*", r"<strong>\1</strong>", part)
-        part = re.sub(r"\[([^\]]+)\]\(([^)\s]+)\)", r'<a href="\2">\1</a>', part)
+        part = re.sub(r"\[([^\]]+)\]\(([^)\s]+)\)", _link, part)
         out.append(part)
     return "".join(out)
 
 
+# A scheme before any slash: javascript:, data:, vbscript: and the like.
+_SCHEME = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:")
+
+
+def _link(match: re.Match) -> str:
+    """A Markdown link as an anchor when its target is http(s) or relative,
+    else left as the text it was. The target is already escaped for text,
+    quotes are escaped for the attribute."""
+    label, href = match.group(1), match.group(2)
+    if _SCHEME.match(href) and not re.match(r"^https?://", href, re.IGNORECASE):
+        return match.group(0)
+    return f'<a href="{href.replace(chr(34), "&quot;").replace(chr(39), "&#x27;")}">{label}</a>'
+
+
 def _indented(lines: list[str], pad: str) -> list[str]:
     return [f"{pad}{line}" if line.strip() else "" for line in lines]
 

scripts/sitemeasurements.py

@@ -214,6 +214,130 @@ def campaign_summary(root: Path, stored: list[mr.Stored], source: dict | None) -
             "open": open_items}
 
 
+def pack_summary(root: Path, source: dict | None) -> list[dict]:
+    """The latest pack probe of each platform, install path, system, core
+    and game, with a link to its line."""
+    latest: dict[tuple, tuple[int, dict]] = {}
+    for number, line in enumerate(mr.read_campaign(root, mr.PACK_PROBES), 1):
+        key = (line["platform"], line["install"], line["system"], line["core"]["file"],
+               line["content"]["label"])
+        latest[key] = (number, line)
+    items = []
+    for key in sorted(latest):
+        number, line = latest[key]
+        url = (permalink(source["repo"], source["commit"], mr.records_path(mr.PACK_PROBES), number)
+               if source and source.get("commit") else None)
+        items.append({
+            "platform": line["platform"], "platform_version": line["platform_version"],
+            "install": line["install"], "system": line["system"], "emulator": line["emulator"],
+            "core": line["core"]["file"].removesuffix("_libretro.so"),
+            "content": line["content"]["label"], "verdict": line["verdict"],
+            "date": line["probed_at"], "opened": len(line["files_opened"]),
+            "findings": len(line["findings"]), "id": line["id"], "record_url": url,
+            "files": [f"{f['path']} ({f['entry'] or 'written by the core'})" for f in line["files_opened"]],
+            "finding_list": [f"{f['kind']} {f['path']}" for f in line["findings"]],
+            "detail": (line.get("verdict_evidence") or {}).get("detail") or line.get("error") or "",
+        })
+    return items
+
+
+def by_platform(index: dict) -> dict[str, list[dict]]:
+    out: dict[str, list[dict]] = {}
+    for item in index.get("packs") or ():
+        out.setdefault(item["platform"], []).append(item)
+    return out
+
+
+INSTALL_WORDS = {"pack": "pack", "installer": "installer"}
+
+
+def _pack_row(item: dict) -> str:
+    link = f"[line]({item['record_url']})" if item["record_url"] else _code(item["id"])
+    cells = [_plain(item["platform"]), _plain(item["system"]), INSTALL_WORDS[item["install"]],
+             _code(item["core"]), _plain(item["content"]), f"`{item['verdict']}`",
+             str(item["findings"]), link]
+    return "| " + " | ".join(cells) + " |"
+
+
+def packs_section(items: list[dict]) -> list[str]:
+    """The Measurements page's table of packs tested the way a user installs
+    them: one row per platform, system, install path, core and game."""
+    if not items:
+        return []
+    columns = ["Platform", "System", "Installed by", "Core", "Game", "Verdict", "Findings", "Record"]
+    return [
+        "## Packs",
+        "",
+        "A game started with the platform's BIOS folder as a user gets it, from the pack or "
+        f"from the installer ([how]({METHOD_PAGE}#packs)). A finding is a file the core looked "
+        "for that the folder lacks or holds under another path.",
+        "",
+        "| " + " | ".join(columns) + " |",
+        "|" + "|".join("---" for _ in columns) + "|",
+        *(_pack_row(item) for item in
+          sorted(items, key=lambda i: (i["platform"], i["system"], i["install"], i["core"], i["content"]))),
+        "",
+    ]
+
+
+def _inert(value: object) -> str:
+    """Run output (a core's log line, a file name it opened) placed in raw
+    HTML as text: every character HTML, MDX or Markdown reads as markup,
+    quotes included, written as a character reference."""
+    return _plain(value).replace('"', """).replace("'", "'")
+
+
+def _pack_details(item: dict) -> list[str]:
+    def line(label: str, value: object) -> str:
+        return _sentence_html(label, _inert(value), markup_ready=True)
+
+    lines = [
+        line("Result", item["detail"]),
+        line("Core", f"{item['core']} ({item['emulator']})"),
+        line("Files the core opened", ", ".join(item["files"]) or "none"),
+        line("Findings", ", ".join(item["finding_list"]) or "none"),
+        f'<p><a href="{_attr(item["record_url"])}">Full record</a>.</p>' if item["record_url"] else "",
+    ]
+    return [line for line in lines if line]
+
+
+def pack_tested_section(items: list[dict]) -> list[str]:
+    """The Tested block of a platform page, in the emulator pages' Tested
+    markup: one summary line, then, folded under one line as the emulator
+    pages fold their earlier tests, a row per game, core and install path
+    with its verdict chip and date, the files and findings in its details."""
+    if not items:
+        return []
+    works = sum(1 for i in items if i["verdict"] == "boots")
+    last = max(i["date"][:10] for i in items)
+    summary = (f"{len(items)} game start{'s' if len(items) != 1 else ''} with this BIOS folder as "
+               f"you install it · {works} work · last run {last}.")
+    links = (f'<a href="{markup.site_href(METHOD_PAGE.removesuffix(".md") + "/")}#packs">How</a> · '
+             f'<a href="{markup.site_href(EXPLORER_PAGE.removesuffix(".md") + "/")}">All tests</a>')
+    lines = ["## Tested", "", '<div class="rb-tested">',
+             f'<p class="rb-tested-summary">{html.escape(summary)} {links}</p>',
+             '<details class="rb-tested-earlier">',
+             "<summary>Each start, by system</summary>",
+             '<div class="rb-tested-head" aria-hidden="true"><span>Test</span>'
+             "<span>Result</span><span>Installed by</span><span>Date</span></div>"]
+    for item in sorted(items, key=lambda i: (i["system"], i["content"], i["core"], i["install"])):
+        title = (f"{_inert(item['system'])}, {_inert(short_content(item['content']))} "
+                 f"<small>{_inert(item['core'])}</small>")
+        lines += [
+            '<details class="rb-tested-row">',
+            ("<summary>"
+             f'<span class="rb-tested-test">{title}</span>'
+             f'<span class="rb-tested-result">{_chip({"status": item["verdict"]})}</span>'
+             f'<span class="rb-tested-variants">{INSTALL_WORDS[item["install"]]}</span>'
+             f'<span class="rb-tested-date">{item["date"][:10]}</span></summary>'),
+            '<div class="rb-tested-more">',
+            *_pack_details(item),
+            "</div>",
+            "</details>",
+        ]
+    return [*lines, "</details>", "</div>", ""]
+
+
 # Characters a campaign value could use to leave its place on the page: HTML
 # and MDX syntax ({ } opens an expression), code spans, emphasis and links.
 _UNSAFE = {"&": "&amp;", "<": "&lt;", ">": "&gt;", "{": "&#123;", "}": "&#125;", "`": "&#96;",
@@ -682,6 +806,7 @@ def build(public: Path, root: Path = ROOT, emulators_dir: Path = ROOT / "emulato
     index = build_index(stored, source, profiles, system_names)
     index["campaign"] = campaign_summary(root, stored, source)
     index["coverage"] = coverage(root, stored, profiles)
+    index["packs"] = pack_summary(root, source)
     paths = [public / INDEX_PATH]
     (public / INDEX_PATH).parent.mkdir(parents=True, exist_ok=True)
     write_if_changed(str(public / INDEX_PATH),
@@ -706,6 +831,9 @@ def by_emulator(index: dict) -> dict[str, list[dict]]:
         opened[line["emulator"]] = opened.get(line["emulator"], 0) + 1
     out: dict[str, list[dict]] = {}
     for item in index.get("items", []):
+        # A pack probe's run is shown on its platform's page, beside the pack.
+        if item["subject"].startswith(mr.PACK_SUBJECT_PREFIX):
+            continue
         out.setdefault(item["emulator"], []).append(
             {**item, "open_findings": opened.get(item["emulator"], 0)})
     return out
@@ -1186,6 +1314,7 @@ def generate_explorer_page(index: dict, system_names: dict[str, str] | None = No
         "",
         "<MeasurementsExplorer />",
         "",
+        *packs_section(index.get("packs") or []),
         *campaign_section(index.get("campaign")),
         *coverage_section(index.get("coverage"), system_names),
     ])