fix: order inherited rescrapes, end on verify

refresh_stale ran a parent's rescrape and its children's truth and
export jobs side by side, so a child read the parent mid-rewrite. A job
now starts once the rescrapes of its ancestors have ended, and a child
no row names gets its export rewritten after them. A rescrape chain ends
on verify.py --platform, whose summary closes the report.

scripts/refresh_stale.py

@@ -11,7 +11,10 @@ Mapping (stale -> command):
     platforms/<name>        python -m scripts.scraper.<module>_scraper
                             -o platforms/<name>.yml
                             then the native refresh below: the original
-                            the export patches must be the one just scraped
+                            the export patches must be the one just scraped,
+                            then python scripts/verify.py --platform <name>,
+                            whose summary the report prints: a rescrape can
+                            add systems whose files are missing
     native/<name>           python scripts/generate_truth.py --platform <name>, then
                             python scripts/export_native.py --platform <name>
                             --refresh-cache, then the export itself, which
@@ -25,6 +28,10 @@ Mapping (stale -> command):
     catalogs/fbneo recipes  python -m scripts.scraper.romset_dat_importer
                             --source fbneo --fetch
 
+A platform that inherits from a rescraped one (retropie and lakka from
+retroarch) reads the parent's file: its jobs start once the parent's chain
+has ended, and its export is rewritten even when no row named it.
+
 Surfaced only (no refresh): data rows that are not a _data_dirs.yml key
 (the registry itself, buildbot assets it does not track), coreinfo gaps, redump/no-intro/tosec packs,
 CI pins (install.py SHA-256, PyPI, pinned actions), profile_sync. Each needs
@@ -45,7 +52,7 @@ import re
 import subprocess
 import sys
 import time
-from concurrent.futures import ThreadPoolExecutor, as_completed
+from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
 from dataclasses import dataclass
 from functools import lru_cache
 from pathlib import Path
@@ -72,6 +79,8 @@ class Job:
     log_path: Path
     # Commands run after `command`, in order, only while each one succeeds.
     then: tuple[tuple[str, ...], ...] = ()
+    # Platforms whose rescrape job must end before this one starts.
+    after: frozenset[str] = frozenset()
 
 
 @dataclass(frozen=True)
@@ -94,6 +103,30 @@ def _load_platform_registry() -> dict[str, dict]:
         return (yaml.safe_load(fh) or {}).get("platforms") or {}
 
 
+def _platform_parents(registry: dict[str, dict]) -> dict[str, str]:
+    """Platform to the platform its YAML inherits from."""
+    import yaml
+
+    parents: dict[str, str] = {}
+    for name in registry:
+        path = REPO_ROOT / "platforms" / f"{name}.yml"
+        if not path.is_file():
+            continue
+        with path.open(encoding="utf-8") as fh:
+            parent = (yaml.safe_load(fh) or {}).get("inherits")
+        if parent:
+            parents[name] = str(parent)
+    return parents
+
+
+def _ancestors(platform: str, parents: dict[str, str]) -> set[str]:
+    found: set[str] = set()
+    while platform in parents and parents[platform] not in found:
+        platform = parents[platform]
+        found.add(platform)
+    return found
+
+
 @lru_cache(maxsize=1)
 def _data_dir_keys() -> frozenset[str]:
     """The entries refresh_data_dirs.py --key accepts."""
@@ -129,15 +162,21 @@ def _slug(subject: str) -> str:
 
 
 def plan_jobs(
-    findings: list[dict], registry: dict[str, dict]
+    findings: list[dict],
+    registry: dict[str, dict],
+    parents: dict[str, str] | None = None,
 ) -> tuple[list[Job], list[dict], list[dict]]:
     """Split findings into (refreshable jobs, surfaced-only, ignored).
 
     Surfaced-only entries carry stale state that only a human can resolve
     (new .info file, manual DAT download, PyPI bump). Ignored entries are
     anything not STALE (OK, SKIPPED, UNKNOWN, ERROR). ERROR is also surfaced
-    so the final report mentions it.
+    so the final report mentions it. ``parents`` maps a platform to the one
+    it inherits from: a child's jobs wait for its rescraped ancestors, and a
+    child no row names gets its export rewritten after them.
     """
+    if parents is None:
+        parents = _platform_parents(registry)
     jobs: list[Job] = []
     seen: set[tuple[str, str]] = set()
     surfaced: list[dict] = []
@@ -189,16 +228,92 @@ def plan_jobs(
                 tuple(_truth_refresh(subject)),
                 tuple(_native_refresh(subject)),
                 tuple(_export_refresh(subject)),
+                tuple(_verify(subject)),
             )
         elif area == "native":
             then = (tuple(_native_refresh(subject)), tuple(_export_refresh(subject)))
+        after: frozenset[str] = frozenset()
+        if area in ("platforms", "native"):
+            after = frozenset(_ancestors(subject, parents) & rescraped)
         jobs.append(
-            Job(area=area, subject=subject, command=command, log_path=log, then=then)
+            Job(area=area, subject=subject, command=command, log_path=log,
+                then=then, after=after)
         )
 
+    # The export of a child is built from its parent's file: rewritten once
+    # the parent is, or the next check reports it stale.
+    covered = {j.subject for j in jobs if j.area in ("platforms", "native")}
+    for child in sorted(parents):
+        after = frozenset(_ancestors(child, parents) & rescraped)
+        if not after or child in covered:
+            continue
+        jobs.append(Job(
+            area="native",
+            subject=child,
+            command=_truth_refresh(child),
+            log_path=LOG_DIR / f"native__{_slug(child)}.log",
+            then=(tuple(_native_refresh(child)), tuple(_export_refresh(child))),
+            after=after,
+        ))
+
     return jobs, surfaced, ignored
 
 
+def _verify(platform: str) -> list[str]:
+    """The check a rescraped platform file is read with before it is kept."""
+    return [sys.executable, "scripts/verify.py", "--platform", platform]
+
+
+VERIFY_MARK = "scripts/verify.py --platform"
+
+
+def verify_summary(path: Path) -> str:
+    """First line verify printed in a job log, empty when it did not run."""
+    try:
+        text = path.read_text(encoding="utf-8", errors="replace")
+    except OSError:
+        return ""
+    lines = text.splitlines()
+    for index, line in enumerate(lines):
+        if line.startswith("$ ") and VERIFY_MARK in line:
+            return next(
+                (rest.strip() for rest in lines[index + 1:]
+                 if rest.strip() and not rest.startswith("$ ")),
+                "",
+            )
+    return ""
+
+
+def run_jobs(jobs: list[Job], env: dict[str, str], workers: int) -> list[JobResult]:
+    """Run jobs in parallel, each once the rescrapes it waits for have ended.
+
+    A failed parent still releases its children: its file is unchanged, and
+    the child reads it as it is.
+    """
+    results: list[JobResult] = []
+    ended: set[str] = set()
+    waiting = list(jobs)
+    with ThreadPoolExecutor(max_workers=max(1, workers)) as pool:
+        running: dict = {}
+        while waiting or running:
+            ready = [job for job in waiting if job.after <= ended]
+            for job in ready:
+                waiting.remove(job)
+                running[pool.submit(run_job, job, env)] = job
+            if not running:
+                raise RuntimeError(
+                    "jobs wait on rescrapes that are not planned: "
+                    + ", ".join(sorted(j.subject for j in waiting))
+                )
+            done, _ = wait(running, return_when=FIRST_COMPLETED)
+            for future in done:
+                job = running.pop(future)
+                results.append(future.result())
+                if job.area == "platforms":
+                    ended.add(job.subject)
+    return results
+
+
 def _truth_refresh(platform: str) -> list[str]:
     return [sys.executable, "scripts/generate_truth.py", "--platform", platform]
 
@@ -380,6 +495,22 @@ def render(
             if r.returncode != 0 and r.tail:
                 lines.append(f"           -> {r.tail}")
 
+        rescraped = sorted(
+            (r for r in results if r.job.area == "platforms" and r.returncode == 0),
+            key=lambda r: r.job.subject,
+        )
+        if rescraped:
+            lines.append(
+                "\n[verify] a rescrape can add systems: read each report "
+                "before keeping platforms/<name>.yml"
+            )
+            for r in rescraped:
+                summary = verify_summary(r.job.log_path) or "(no verify output)"
+                lines.append(f"  {r.job.subject:12} {summary}")
+                lines.append(
+                    f"  {'':12} python scripts/verify.py --platform {r.job.subject}"
+                )
+
     if surfaced:
         lines.append("\n[manual review needed]")
         for f in surfaced:
@@ -457,6 +588,8 @@ def main() -> int:
               f"{len(ignored)} already clean/skipped")
         for job in sorted(jobs, key=lambda j: (j.area, j.subject)):
             print(f"  {job.area:10} {job.subject:28}  {' '.join(job.command)}")
+            if job.after:
+                print(f"  {'':10} {'':28}  after {', '.join(sorted(job.after))}")
             for command in job.then:
                 print(f"  {'':10} {'':28}  then {' '.join(command)}")
         if surfaced:
@@ -468,11 +601,7 @@ def main() -> int:
 
     LOG_DIR.mkdir(parents=True, exist_ok=True)
     env = _github_token_env()
-    results: list[JobResult] = []
-    with ThreadPoolExecutor(max_workers=max(1, args.jobs)) as pool:
-        futures = [pool.submit(run_job, job, env) for job in jobs]
-        for future in as_completed(futures):
-            results.append(future.result())
+    results = run_jobs(jobs, env, args.jobs)
 
     if args.json:
         payload = {

tests/test_refresh_stale.py

@@ -30,8 +30,8 @@ def finding(area: str, subject: str, status: str = "STALE") -> dict:
 
 
 class Planning(unittest.TestCase):
-    def _plan(self, *findings: dict):
-        return rs.plan_jobs(list(findings), REGISTRY)
+    def _plan(self, *findings: dict, parents: dict | None = None):
+        return rs.plan_jobs(list(findings), REGISTRY, parents or {})
 
     def test_a_stale_platform_is_rescraped_then_its_original_is_refreshed(self):
         jobs, surfaced, _ = self._plan(finding("platforms", "retrobat"))
@@ -47,9 +47,28 @@ class Planning(unittest.TestCase):
             [list(c[1:]) for c in jobs[0].then],
             [["scripts/generate_truth.py", "--platform", "retrobat"],
              ["scripts/export_native.py", "--platform", "retrobat", "--refresh-cache"],
-             ["scripts/export_native.py", "--platform", "retrobat"]],
+             ["scripts/export_native.py", "--platform", "retrobat"],
+             ["scripts/verify.py", "--platform", "retrobat"]],
         )
 
+    def test_a_child_waits_for_its_rescraped_parent(self):
+        """retropie inherits retroarch: run beside the parent's rescrape, its
+        truth read the old parent and its export was refused as stale."""
+        registry = dict(REGISTRY, retroarch={"scraper": "libretro"},
+                        retropie={"scraper": "retropie"})
+        parents = {"retropie": "retroarch", "lakka": "retroarch"}
+        jobs, _, _ = rs.plan_jobs(
+            [finding("platforms", "retroarch"), finding("platforms", "retropie"),
+             finding("platforms", "retrobat")],
+            registry, parents,
+        )
+        after = {(j.area, j.subject): j.after for j in jobs}
+        self.assertEqual(after[("platforms", "retroarch")], frozenset())
+        self.assertEqual(after[("platforms", "retrobat")], frozenset())
+        self.assertEqual(after[("platforms", "retropie")], frozenset({"retroarch"}))
+        # lakka has no row and no scraper: its export follows the parent's.
+        self.assertEqual(after[("native", "lakka")], frozenset({"retroarch"}))
+
     def test_a_native_row_of_a_rescraped_platform_is_not_a_second_job(self):
         """Two jobs would race: the refresh must follow the scrape."""
         jobs, _, ignored = self._plan(
@@ -132,6 +151,33 @@ class Planning(unittest.TestCase):
 
 
 class Running(unittest.TestCase):
+    def test_a_child_starts_after_its_parent_ends(self):
+        with tempfile.TemporaryDirectory() as directory:
+            order = Path(directory) / "order"
+            def job(area, subject, after=()):
+                code = f"open({str(order)!r}, 'a').write({subject!r} + chr(10))"
+                if subject == "parent":
+                    code = "import time; time.sleep(0.5); " + code
+                return rs.Job(area, subject, [sys.executable, "-c", code],
+                              Path(directory) / f"{subject}.log", (), frozenset(after))
+            rs.run_jobs([job("native", "child", ["parent"]),
+                         job("platforms", "parent")], {}, 4)
+            self.assertEqual(order.read_text().split(), ["parent", "child"])
+
+    def test_the_report_prints_verify_after_a_rescrape(self):
+        (REPO_ROOT / "tmp").mkdir(exist_ok=True)
+        with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as directory:
+            log = Path(directory) / "x.log"
+            log.write_text(
+                "$ python -m scripts.scraper.x_scraper\nscraped\n"
+                "$ python scripts/verify.py --platform x\n"
+                "X: 10/12 OK, 2 missing [md5]\n"
+            )
+            job = rs.Job("platforms", "x", ["true"], log)
+            text = rs.render([rs.JobResult(job, 0, 1.0, "")], [], [])
+        self.assertIn("X: 10/12 OK, 2 missing [md5]", text)
+        self.assertIn("python scripts/verify.py --platform x", text)
+
     def _job(self, directory: str, command: list[str], then=()) -> rs.Job:
         return rs.Job("platforms", "x", command, Path(directory) / "x.log", then)
 

wiki/tools.md

@@ -611,8 +611,8 @@ committed.
 
 | Stale row | Command |
 |-----------|---------|
-| `platforms/<name>` | the platform scraper, then the `native` refresh for the same platform |
-| `native/<name>` | `export_native.py --platform <name> --refresh-cache` |
+| `platforms/<name>` | the platform scraper, then the `native` refresh for the same platform, then `verify.py --platform <name>` |
+| `native/<name>` | `generate_truth.py --platform <name>`, `export_native.py --platform <name> --refresh-cache`, then the export itself |
 | `targets/<name>` | the target scraper |
 | `data/<key>` | `refresh_data_dirs.py --key <key> --force` |
 | `catalogs/mame recipes`, `catalogs/fbneo recipes` | `scraper/romset_dat_importer.py --source <source> --fetch` |
@@ -623,6 +623,13 @@ pins, and `profile_sync`. A failed scraper leaves its cached original
 untouched. Each job writes its output to `tmp/refresh_stale/`, and a GitHub
 token is taken from `GITHUB_TOKEN` or from `gh auth token`.
 
+A platform that inherits from a rescraped one (`lakka` and `retropie` from
+`retroarch`) reads the parent's file, so its jobs start once the parent's have
+ended, and its export is rewritten even when no row named it. A rescrape can
+add systems whose files the collection lacks: the report ends with the
+`verify.py --platform` summary of every rescraped platform, to read before
+keeping the new YAML.
+
 ```bash
 python scripts/refresh_stale.py --dry-run
 python scripts/refresh_stale.py