chore: merge the remaining launcher proofs

scripts/check_freshness.py

@@ -282,7 +282,8 @@ def diff_targets(old: dict, new: dict) -> PlatformDiff:
 
 
 def _scrape_into_copy(
-    module: str, source: Path, workdir: Path, timeout: int = 900
+    module: str, source: Path, workdir: Path, timeout: int = 900,
+    extra: tuple[str, ...] = (),
 ) -> tuple[Path, str | None]:
     """Run a scraper's own write path on a throwaway copy of *source*.
 
@@ -292,7 +293,7 @@ def _scrape_into_copy(
     target = workdir / source.name
     shutil.copyfile(source, target)
     proc = subprocess.run(
-        [sys.executable, "-m", module, "-o", str(target)],
+        [sys.executable, "-m", module, "-o", str(target), *extra],
         cwd=REPO_ROOT,
         capture_output=True,
         text=True,
@@ -327,6 +328,17 @@ def _scrapable_platforms(platforms_dir: Path) -> list[tuple[str, str, Path]]:
     return rows
 
 
+def rescrape_args(platforms_dir: Path, name: str) -> tuple[str, ...]:
+    """What a rescrape passes so the scraper reads upstream's moving head.
+
+    A scraper that rereads the revision its platform file pins by default
+    (RetroBat) names the ref to move to in the registry's rescrape_args.
+    """
+    with (platforms_dir / "_registry.yml").open(encoding="utf-8") as fh:
+        registry = (yaml_load(fh) or {}).get("platforms") or {}
+    return tuple(str(a) for a in (registry.get(name) or {}).get("rescrape_args") or ())
+
+
 def check_platforms(platforms_dir: Path, workdir: Path, jobs: int) -> list[Finding]:
     rows = _scrapable_platforms(platforms_dir)
     findings: list[Finding] = []
@@ -334,7 +346,9 @@ def check_platforms(platforms_dir: Path, workdir: Path, jobs: int) -> list[Findi
     def run(row: tuple[str, str, Path]) -> Finding:
         name, module, path = row
         try:
-            fresh, error = _scrape_into_copy(module, path, workdir)
+            fresh, error = _scrape_into_copy(
+                module, path, workdir, extra=rescrape_args(platforms_dir, name)
+            )
         except subprocess.TimeoutExpired:
             return Finding("platforms", name, ERROR, detail="scraper timed out")
         if error:

scripts/launchers.py

@@ -369,13 +369,17 @@ def pool_folders(profile: dict) -> dict[str, str]:
 def _reads_apply(read: dict, profile: dict, platform: str) -> bool:
     """Whether a config_reads entry describes the build ``platform`` ships.
 
-    An entry naming platforms is read at their pin_build revision. One that
-    names none is read at the profile's pin, the build of every platform
-    with no pin_build of its own or one pinning the same commit.
+    An entry naming platforms is read at their pin_build revision or its
+    revision. One that names none is read at the profile's pin, the build
+    of every platform no entry names and with no pin_build of its own or
+    one pinning the same commit.
     """
     named = [str(p) for p in read.get("platforms") or []]
     if named:
         return platform in named
+    # A platform an entry names has its build read there, not at the pin.
+    if any(platform in (r.get("platforms") or []) for r in profile.get("config_reads") or []):
+        return False
     pin = str(profile.get("source_commit") or "")
     builds = [b for b in profile.get("pin_build") or [] if b.get("platform") == platform]
     return all(pin and str(b.get("value") or "").startswith(pin[:12]) for b in builds)
@@ -468,7 +472,12 @@ def config_evidence(
     readers: list[str] = []
     uncovered = list(keys)
     loaded = False
-    for key in _named_profiles(record, profiles, platform) or sorted(profiles):
+    named = _named_profiles(record, profiles, platform)
+    if file == CORE_OPTIONS:
+        # A core option is read by the core RetroArch runs, not by the
+        # frontend a record names (RetroDECK's retroarch component).
+        named = [k for k in named if "libretro" in str(profiles[k].get("type") or "")]
+    for key in named or sorted(profiles):
         profile = profiles[key]
         handles = _handle_refs(profile)
         for read in profile.get("config_reads") or []:
@@ -1319,7 +1328,10 @@ class LauncherLayer:
                 for key in self._profiles_named(emulator, profiles, relevant)
             }
             # RetroDECK names its frontend, retroarch, and the core option it
-            # writes, pcsx2_bios, names the core.
+            # writes, pcsx2_bios, names the core: a libretro core option
+            # belongs to a libretro core.
+            if place.build == "libretro" and place.kind == "core_option":
+                owners = {k for k in owners if "libretro" in str(profiles[k].get("type") or "")}
             if not owners:
                 owners = set(setting_readers(place.passes_to, profiles, relevant))
             for key in sorted(owners):

scripts/profile_sync.py

@@ -1652,7 +1652,10 @@ def _check_build_refs(
     results: list[dict] = []
     for citation, read in deferred:
         for platform in read.get("platforms") or []:
-            revision = str(read.get("revision") or "") or build_revision(profile, str(platform))
+            # A closed build is named by its binary: an address ref is
+            # skipped below, a source path cannot exist at a sha256.
+            revision = (str(read.get("revision") or read.get("binary_sha256") or "")
+                        or build_revision(profile, str(platform)))
             row = {"platform": str(platform), "revision": revision or "",
                    "ref": citation.ref, "field": citation.field}
             if revision is None:

scripts/refresh_stale.py

@@ -380,6 +380,7 @@ def _command_for(
             f"scripts.scraper.{scraper}_scraper",
             "-o",
             f"platforms/{subject}.yml",
+            *(str(a) for a in entry.get("rescrape_args") or ()),
         ]
 
     if area == "targets":

scripts/scraper/base_scraper.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import inspect
 import json
 import re
 import sys
@@ -391,10 +392,16 @@ def scraper_cli(
     parser.add_argument("--dry-run", action="store_true", help="Show scraped data")
     parser.add_argument("--output", "-o", help="Output YAML file")
     parser.add_argument("--json", action="store_true", help="Output as JSON")
+    takes_revision = "revision" in inspect.signature(scraper_class).parameters
+    if takes_revision:
+        parser.add_argument(
+            "--revision",
+            help="Read this branch head or commit instead of the revision the platform file pins",
+        )
     args = parser.parse_args()
 
     try:
-        scraper = scraper_class()
+        scraper = scraper_class(revision=args.revision) if takes_revision else scraper_class()
         reqs = scraper.fetch_requirements()
     except (OSError, ValueError, RuntimeError) as e:
         _exit_on_scrape_error(e)

scripts/scraper/emuelec_launchers.py

@@ -40,6 +40,10 @@ EMULATORS = f"{SX05RE}/emulators"
 LIBRETRO = f"{SX05RE}/libretro"
 RETROARCH_MK = f"{SX05RE}/libretro_base/retroarch/package.mk"
 BIOS_ROOT = "/storage/roms/bios"
+EMUELEC_MK = f"{SX05RE}/emuelec/package.mk"
+# The copy of every package's /usr/config into /storage/.config at each boot,
+# an existing file kept (cp -i answered by false).
+USERCONFIG = "packages/sysutils/systemd/scripts/userconfig-setup"
 # The roms root that holds it (packages/sx05re/libretro_base/retroarch/package.mk:112,117).
 ROMS_ROOT = "/storage/roms"
 
@@ -106,6 +110,19 @@ class Runs:
         return (RUN_EMU, set(self.of(name).lines))
 
 
+def _storage_path(tree: RepoTree, target: str) -> str:
+    """target with the /emuelec link resolved to the folder it points at.
+
+    The emuelec package links /emuelec to /storage/.config/emuelec, where
+    the first-boot copy of /usr/config lands.
+    """
+    mk = _read(tree, EMUELEC_MK)
+    _line, link = mk.find(r"ln -sf (/storage/\S+) \$\{INSTALL\}(/emuelec)$", "/emuelec link")
+    if target == link.group(2) or target.startswith(link.group(2) + "/"):
+        return link.group(1) + target[len(link.group(2)):]
+    return target
+
+
 def _evidence(
     record: Launcher,
     script: Script,
@@ -113,15 +130,28 @@ def _evidence(
     keys: list[str],
     source_name: str = "",
     rewrites: Iterable[int] = (),
+    tree: RepoTree | None = None,
+    installed: Iterable[tuple[str, Iterable[int]]] = (),
 ) -> Launcher:
-    """A config record with the start script lines that put its file in place.
+    """A config record with the lines that put its file in place: the
+    first-boot copy of /usr/config when ``tree`` is given, then the start
+    script's own. ``installed`` are the package.mk lines putting the file
+    in /usr/config.
 
     ``rewrites`` are lines the extractor read as rewriting the record's
     values without naming their keys (advmame.sh's sed of /roms/mame).
     """
-    placement = config_placement(script, target, keys, source_name=source_name)
+    placements = []
+    if tree is not None:
+        setup = _read(tree, USERCONFIG)
+        placements.append(config_placement(
+            setup, _storage_path(tree, target), required=False, sources=("/usr/config/",)
+        ))
+    placement = config_placement(script, target, keys, source_name=source_name,
+                                 required=not placements or not placements[0].install)
     placement.overrides |= set(rewrites)
-    fields = evidence_fields([placement])
+    placements.append(placement)
+    fields = evidence_fields(placements, installed)
     record.installs = fields["installs"]
     record.kept_if_present = fields["kept_if_present"]
     record.overrides = fields["overrides"]
@@ -356,6 +386,8 @@ def duckstation(tree: RepoTree, runs: Runs) -> list[Launcher]:
         script,
         script.expand("${CONFIG_DIR}/settings.ini", run_line, set()),
         [f"[{s}] {k}" for s, k in (folder_key, *file_keys)],
+        tree=tree,
+        installed=[(mk.path, {mk_line})],
     )]
 
 
@@ -569,7 +601,8 @@ def advancemame(tree: RepoTree, runs: Runs) -> list[Launcher]:
     target = script.expand("${CONFIG_DIR}/advmame.rc", run_line, set())
 
     def proved(record: Launcher, key: str) -> Launcher:
-        return _evidence(record, script, target, [key], rewrites={sed_line})
+        return _evidence(record, script, target, [key], rewrites={sed_line}, tree=tree,
+                         installed=[(mk.path, {mk_line for _, _, mk_line in configs})])
 
     records = []
     for key, folder in value:

scripts/scraper/retrobat_launchers.py

@@ -1868,13 +1868,27 @@ def _root_writes(src: Source, emulator: str) -> list[Record]:
         if root_reach is None:
             continue
         method = root_reach[0]
+        # A folder below the root, `string x = Path.Combine(root, "cdb")`
+        # that the method creates or tests as a directory. A file below it
+        # is a guarded write's, read by _guarded_write.
+        folders = {root.group(1): ("bios", root)}
+        for sub in re.compile(
+            rf"string (\w+) = Path\.Combine\({root.group(1)}, ((?:{STRING}, )*{STRING})\);"
+        ).finditer(src.text, root.end(), method[1]):
+            if re.search(
+                rf"\b(?:\w*CreateDirectory|Directory\.Exists)\({sub.group(1)}\)",
+                src.text[sub.end() : method[1]],
+            ):
+                folders[sub.group(1)] = ("/".join(["bios", *_strings(sub.group(2))]), sub)
+        names = "|".join(map(re.escape, folders))
         for write in re.compile(
-            rf'(\w+)\.WriteValue\("([^"]*)", "([^"]+)", (?:"\'" \+ )?{root.group(1)}(?: \+ "\'")?\);'
+            rf'(\w+)\.WriteValue\("([^"]*)", "([^"]+)", (?:"\'" \+ )?({names})(?: \+ "\'")?\);'
         ).finditer(src.text, root.end(), method[1]):
             reached = _conditions(src, write.start())
             if reached is None:
                 continue
-            writer, section, key = write.groups()
+            writer, section, key, held = write.groups()
+            folder, declared = folders[held]
             target = f"[{section}] {key}" if section else key
             records.append(
                 _record(
@@ -1888,12 +1902,14 @@ def _root_writes(src: Source, emulator: str) -> list[Record]:
                             [
                                 *reached[2],
                                 src.lines(root.start(), root.end()),
+                                *([src.lines(declared.start(), declared.end())]
+                                  if declared is not root else []),
                                 src.lines(write.start(), write.end()),
                             ],
                         )
                     ),
-                    search_dir="bios",
-                    passes_to=f"{_config_name(src, method, writer)} {target} = bios",
+                    search_dir=folder,
+                    passes_to=f"{_config_name(src, method, writer)} {target} = {folder}",
                     fails_when_missing=False,
                     depends_on_option=", ".join(reached[1]) or None,
                 )
@@ -2376,11 +2392,7 @@ def _option_path_in(src: Source, features: Source, emulator: str) -> list[Record
     gate = src.find(
         rf'if \(SystemConfig\.isOptSet\("{option}"\)', f"{option} gate", method[0], override.start()
     )
-    gate_block = src.block_at(gate.end())
-    if not gate_block[0] < override.start() < gate_block[1]:
-        raise LauncherPatternMissing(
-            src.path, f"{option} block around line {src.line(override.start())}"
-        )
+    gate_block = _gate_span(src, gate, override)
     write = re.compile(
         rf'(\w+)\.WriteValue\("([^"]*)", "([^"]+)", ([^;\n]*\b{var}\b[^;\n]*)\);'
     ).search(code, override.end(), method[1])
@@ -2390,6 +2402,8 @@ def _option_path_in(src: Source, features: Source, emulator: str) -> list[Record
     guarded = re.compile(rf"if \(File\.Exists\({var}\)[^\n]*\)\s*$").search(
         code, method[0], write.start()
     )
+    flag = None if guarded else _flag_guard(code, var, method[0], write.start())
+    guarded = guarded or flag
     skipped = re.findall(
         rf'SystemConfig\["{option}"\] != "(\w+)"', code[gate.start() : write.end()]
     )
@@ -2397,8 +2411,13 @@ def _option_path_in(src: Source, features: Source, emulator: str) -> list[Record
     target = f"{config} [{section}] {key}" if section else f"{config} {key}"
     suffix = ", only if the file exists" if guarded else ", written without an existence check"
     fails = f"{key} not written" if guarded else "path written without an existence check"
-    write_span = src.lines(_statement_start(src, write.start()), write.end())
+    write_span = src.lines(
+        flag.start() if flag else _statement_start(src, write.start()), write.end()
+    )
     choices, choice_lines = feature_choices(features, emulator, option)
+    fallback = _fallback_names(src, var, gate_block[1], write.start())
+    if fallback is not None:
+        fails = f"{', then '.join(fallback[0])}, instead, else {fails}"
     records = []
     default = list(
         re.compile(rf'string {var} = Path\.Combine\({_BIOS_ROOT}, "([^"]+)"\);').finditer(
@@ -2437,9 +2456,71 @@ def _option_path_in(src: Source, features: Source, emulator: str) -> list[Record
             depends_on_option=option,
         )
     )
+    if fallback is not None:
+        names, span = fallback
+        unset = f"{default[-1].group(1)} absent with {option} unset" if default else None
+        chosen = f"the {option} file absent" + (
+            f", {option} not {' or '.join(skipped)}" if skipped else ""
+        )
+        records.append(
+            _record(
+                emulator,
+                "standalone",
+                "names_searched",
+                names,
+                _path_parts(src, span, write_span),
+                case_insensitive=True,
+                search_dir="bios",
+                passes_to=f"{target} = <first found>{suffix}",
+                fails_when_missing=f"{key} not written" if guarded else fails,
+                depends_on_option=" or ".join(c for c in (unset, chosen) if c),
+            )
+        )
     return records + _root_writes(src, emulator)
 
 
+def _gate_span(src: Source, gate: re.Match[str], override: re.Match[str]) -> tuple[int, int]:
+    """The span an option gate governs: its block, or the one statement a
+    braceless if runs, which must be the override."""
+    paren = gate.start() + len("if ")
+    after = paren + len(_call_args(src, paren)) + 2
+    if re.match(r"\s*\{", src.text[after:]):
+        block = src.block_at(after)
+    else:
+        block = (after, _statement_end(src, after))
+    if not block[0] < override.start() < block[1]:
+        raise LauncherPatternMissing(
+            src.path, f"option block around line {src.line(override.start())}"
+        )
+    return block
+
+
+def _flag_guard(code: str, var: str, start: int, end: int) -> re.Match[str] | None:
+    """A write under `if (flag)` where flag holds File.Exists of the path."""
+    flag = re.compile(rf"bool (\w+) = [^;\n]*\bFile\.Exists\({var}\)[^;\n]*;").search(
+        code, start, end
+    )
+    if flag is None or not re.compile(rf"if \({flag.group(1)}\)\s*$", re.M).search(
+        code, flag.end(), end
+    ):
+        return None
+    return flag
+
+
+def _fallback_names(
+    src: Source, var: str, start: int, end: int
+) -> tuple[list[str], tuple[int, int]] | None:
+    """Names tried in bios, first existing one kept, when the chosen file is absent."""
+    found = re.compile(
+        rf"if \(([^\n]*!File\.Exists\({var}\)[^\n]*)\)\s*"
+        rf"{var} = new\[\] \{{ (?P<names>(?:{STRING}(?:, )?)+) \}}\.Select\((?P<arg>\w+) => "
+        rf"Path\.Combine\({_BIOS_ROOT}, (?P=arg)\)\)\.FirstOrDefault\(File\.Exists\) \?\? {var};"
+    ).search(_code(src), start, end)
+    if found is None:
+        return None
+    return _strings(found.group("names")), src.lines(found.start(), found.end())
+
+
 OPTION_PATHS = {
     f"{GENERATORS}/Kronos.Generator.cs": "kronos",
     f"{GENERATORS}/SSF.Generator.cs": "ssf",

scripts/scraper/retrobat_scraper.py

@@ -12,6 +12,7 @@ import json
 import re
 import urllib.error
 import urllib.request
+from pathlib import Path
 
 try:
     from .base_scraper import (
@@ -39,6 +40,7 @@ except ImportError:
     from retrobat_launchers import extract_launchers
 
 PLATFORM_NAME = "retrobat"
+PLATFORM_FILE = Path(__file__).resolve().parents[2] / "platforms" / "retrobat.yml"
 
 LAUNCHER_REPO = "RetroBat-Official/emulatorlauncher"
 SOURCE_URL = (
@@ -132,6 +134,28 @@ def launcher_revision(url: str) -> str:
     return found.group(1)
 
 
+def pinned_revision(platform_file: Path = PLATFORM_FILE) -> str:
+    """The emulatorLauncher commit platforms/retrobat.yml was read from.
+
+    A rescrape reads this commit unless told to move, so a change to an
+    extractor shows in the diff alone, without upstream's commits since.
+    """
+    from common import yaml_load
+
+    with platform_file.open(encoding="utf-8") as fh:
+        source = str((yaml_load(fh) or {}).get("source") or "")
+    return launcher_revision(source)
+
+
+def resolve_revision(revision: str | None) -> str:
+    """The commit to read: the pin when None, a branch's head, or a sha."""
+    if revision is None:
+        return pinned_revision()
+    if re.fullmatch(r"[0-9a-f]{40}", revision):
+        return revision
+    return fetch_branch_head(LAUNCHER_REPO, revision)
+
+
 def fetch_launchers(rev: str) -> list[dict]:
     """Launcher records read from the generators at the commit the BIOS list comes from."""
     texts = {
@@ -160,10 +184,14 @@ SYSTEM_SLUG_MAP = {
 class Scraper(BaseScraper):
     """Scraper for RetroBat batocera-systems.json."""
 
-    def __init__(self, url: str | None = None) -> None:
-        # batocera-systems.json is read from master: the commit read is recorded.
+    def __init__(self, url: str | None = None, revision: str | None = None) -> None:
+        # The BIOS list and the generators are read at one commit, the one
+        # platforms/retrobat.yml pins unless a revision moves it (--revision
+        # master reads the branch head). The commit read is recorded.
         if url is None:
-            url = SOURCE_URL.format(rev=fetch_branch_head(LAUNCHER_REPO, "master"))
+            url = SOURCE_URL.format(rev=resolve_revision(revision))
+        elif revision is not None:
+            raise ValueError("pass a url or a revision, not both")
         super().__init__(url=url)
         self._parsed: dict | None = None
 

scripts/scraper/shell_launchers.py

@@ -483,6 +483,15 @@ def _guarded(script: Script, number: int, target: str) -> bool:
     return False
 
 
+def _quoted_glob(command: str) -> bool:
+    """Whether the source of a copy holds a "*" only inside quotes."""
+    found = [w for w in words(command)[1:] if not w.raw.startswith("-")]
+    if len(found) < 2:
+        return False
+    raw = found[0].raw
+    return "*" in raw and "*" not in re.sub(r'"[^"]*"|\'[^\']*\'', "", raw)
+
+
 def config_placement(
     script: Script,
     target: str,
@@ -511,12 +520,16 @@ def config_placement(
                 args = []
                 for word in words(line[copy.start(1):])[1:]:
                     # The command ends at a separator or a redirection.
-                    if word.raw in {";", "&&", "||", "|"} or re.match(r"\d*>", word.raw):
+                    if word.raw in {";", "&&", "||", "|"} or re.match(r"\d*>|&>", word.raw):
                         break
                     args.append(word.text)
                 operands = [a for a in args if not a.startswith("-")]
                 if not operands:
                     continue
+                # A quoted "*" is a file named *, which no package installs:
+                # the copy does nothing (EmuELEC's duckstation.sh).
+                if _quoted_glob(line[copy.start(1):]):
+                    continue
                 destination = script.expand(operands[-1], number, set()).rstrip("/")
                 source = script.expand(operands[0], number, set()).rstrip("/")
                 hits = target == destination or target.startswith(destination + "/") or (
@@ -536,7 +549,11 @@ def config_placement(
                     continue
                 if sources and not source.startswith(sources):
                     continue
-                skip = "-n" in args or "--ignore-existing" in args
+                # cp -i fed by false answers no to every overwrite.
+                prompted = any(
+                    a.startswith("-") and not a.startswith("--") and "i" in a[1:] for a in args
+                ) and re.search(r"\bfalse\s*\|\s*$", line[:copy.start(1)])
+                skip = "-n" in args or "--ignore-existing" in args or bool(prompted)
             else:
                 skip = False
             install.add(number)

scripts/validate_schemas.py

@@ -147,7 +147,7 @@ def _config_reads_without_build(path: Path, document: object) -> list[str]:
     for index, read in enumerate(document.get("config_reads") or []):
         if not isinstance(read, dict):
             continue
-        if read.get("revision"):
+        if read.get("revision") or read.get("binary_sha256"):
             continue
         for platform in read.get("platforms") or []:
             if str(platform) not in builds: