refactor: move flag refusals out of the main functions

scripts/generate_pack.py

@@ -2389,6 +2389,75 @@ def _run_platform_packs(
         sys.exit(1)
 
 
+def _refuse_unapplied_flags(args, parser) -> None:
+    """Refuse every flag the requested mode would not apply.
+
+    A mode applies a narrowing flag or refuses it: a swallowed flag answers
+    about an artifact the caller did not name. Run before any quick-exit
+    mode, since --verify-packs and --manifest-targets return early.
+    """
+    # Parsed before the quick-exit modes: --verify-packs returns early and
+    # still needs the region priority list to narrow its expectation.
+    args.regions = []
+    if args.region:
+        try:
+            args.regions = region_mod.parse_requested(args.region)
+        except ValueError as exc:
+            parser.error(str(exc))
+        if args.manifest_targets:
+            parser.error("--region is incompatible with --manifest-targets")
+    if args.one_per_slot and args.manifest_targets:
+        parser.error("--one-per-slot is incompatible with --manifest-targets")
+
+    # --all-variants builds the six source x required combinations itself:
+    # a --source or --required-only beside it would be overridden, and the
+    # modes that build one pack (emulator, system, hashes) never read it.
+    if args.all_variants:
+        for flag, given in (
+            ("--source", args.source != "full"),
+            ("--required-only", args.required_only),
+            ("--emulator", args.emulator),
+            ("--from-md5", args.from_md5 or args.from_md5_file),
+        ):
+            if given:
+                parser.error(f"{flag} is incompatible with --all-variants")
+        if args.system and not (args.platform or args.all):
+            parser.error("--all-variants requires --platform or --all")
+        if args.verify_packs and args.manifest:
+            parser.error("--verify-packs is incompatible with --manifest")
+        if args.verify_packs:
+            # The check that follows the build reads the full pack's name.
+            for flag, given in (
+                ("--system", args.system),
+                ("--one-per-slot", args.one_per_slot),
+                ("--split", args.split),
+            ):
+                if given:
+                    parser.error(f"{flag} is incompatible with --verify-packs")
+
+    if args.verify_packs and not args.all_variants:
+        # Checks packs already on disk against the platform's own list,
+        # narrowed by region and target; it narrows by nothing else.
+        for flag, given in (
+            ("--one-per-slot", args.one_per_slot),
+            ("--required-only", args.required_only),
+            ("--source", args.source != "full"),
+            ("--system", args.system),
+        ):
+            if given:
+                parser.error(f"{flag} is incompatible with --verify-packs")
+    if args.manifest_targets:
+        # One manifest per hardware target, from the target files themselves:
+        # it reads none of the narrowing flags.
+        for flag, given in (
+            ("--target", args.target),
+            ("--required-only", args.required_only),
+            ("--source", args.source != "full"),
+        ):
+            if given:
+                parser.error(f"{flag} is incompatible with --manifest-targets")
+
+
 def main():
 
     parser = argparse.ArgumentParser(description="Generate platform BIOS ZIP packs")
@@ -2493,74 +2562,15 @@ def main():
     args = parser.parse_args()
     packresolve.set_offline(bool(args.offline))
 
-    # Parsed before the quick-exit modes: --verify-packs returns early and
-    # still needs the region priority list to narrow its expectation.
-    args.regions = []
-    if args.region:
-        try:
-            args.regions = region_mod.parse_requested(args.region)
-        except ValueError as exc:
-            parser.error(str(exc))
-        if args.manifest_targets:
-            parser.error("--region is incompatible with --manifest-targets")
-    if args.one_per_slot and args.manifest_targets:
-        parser.error("--one-per-slot is incompatible with --manifest-targets")
-
-    # --all-variants builds the six source x required combinations itself:
-    # a --source or --required-only beside it would be overridden, and the
-    # modes that build one pack (emulator, system, hashes) never read it.
-    if args.all_variants:
-        for flag, given in (
-            ("--source", args.source != "full"),
-            ("--required-only", args.required_only),
-            ("--emulator", args.emulator),
-            ("--from-md5", args.from_md5 or args.from_md5_file),
-        ):
-            if given:
-                parser.error(f"{flag} is incompatible with --all-variants")
-        if args.system and not (args.platform or args.all):
-            parser.error("--all-variants requires --platform or --all")
-        if args.verify_packs and args.manifest:
-            parser.error("--verify-packs is incompatible with --manifest")
-        if args.verify_packs:
-            # The check that follows the build reads the full pack's name.
-            for flag, given in (
-                ("--system", args.system),
-                ("--one-per-slot", args.one_per_slot),
-                ("--split", args.split),
-            ):
-                if given:
-                    parser.error(f"{flag} is incompatible with --verify-packs")
+    _refuse_unapplied_flags(args, parser)
 
     # Quick-exit modes: --verify-packs alone = verify existing packs only
     # Combined with --all-variants, generation runs first then verify
     if args.verify_packs and not args.all_variants:
-        # This mode checks packs already on disk against the platform's own
-        # list, narrowed by the region priority above and by --target. A
-        # narrowing flag it cannot honour is refused rather than dropped: a
-        # dropped flag answers about an artifact the caller did not name.
-        for flag, given in (
-            ("--one-per-slot", args.one_per_slot),
-            ("--required-only", args.required_only),
-            ("--source", args.source != "full"),
-            ("--system", args.system),
-        ):
-            if given:
-                parser.error(f"{flag} is incompatible with --verify-packs")
         with _pack_output_lock(args.output_dir, exclusive=False):
             _run_verify_packs(args)
         return
     if args.manifest_targets:
-        # This mode writes one manifest per hardware target from the target
-        # files themselves; it reads none of the narrowing flags. --region and
-        # --one-per-slot are refused above, and these three the same way.
-        for flag, given in (
-            ("--target", args.target),
-            ("--required-only", args.required_only),
-            ("--source", args.source != "full"),
-        ):
-            if given:
-                parser.error(f"{flag} is incompatible with --manifest-targets")
         with _pack_output_lock(args.output_dir):
             generate_target_manifests(
                 os.path.join(args.platforms_dir, "targets"), args.output_dir

scripts/pipeline.py

@@ -214,6 +214,12 @@ class _Skipped:
 SKIPPED = _Skipped()
 
 
+def _refuse_unapplied_flags(parser: argparse.ArgumentParser, args) -> None:
+    """--all-variants builds every source itself: a --source beside it is dropped."""
+    if args.all_variants and args.source != "full":
+        parser.error("--source is incompatible with --all-variants")
+
+
 def main():
     parser = argparse.ArgumentParser(description="Run the full retrobios pipeline")
     parser.add_argument(
@@ -258,8 +264,7 @@ def main():
         help="Export native formats (implies --with-truth)",
     )
     args = parser.parse_args()
-    if args.all_variants and args.source != "full":
-        parser.error("--source is incompatible with --all-variants")
+    _refuse_unapplied_flags(parser, args)
 
     # A second run on the same output directory is refused before any work:
     # the database rebuild alone takes minutes, and the reader holding the
@@ -460,7 +465,7 @@ def main():
         "3b/8 slot arbitration",
     )
     results["slots"] = ok
-    all_ok = all_ok and ok
+    all_ok &= ok
 
     # Step 4: Generate packs
     pack_output = ""

scripts/profile_sync.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 import argparse
 import contextlib
 import difflib
+import itertools
 import json
 import os
 import posixpath
@@ -1256,6 +1257,28 @@ def detect_pinned_tag(
     return None
 
 
+def _resolve_project_words(refs: list, tree_order) -> list:
+    """Point `project file` parts at the declared tree that vendors them.
+
+    tree_order() yields the trees to search, pin before HEAD; it is only
+    called for a part that needs it, since reading a HEAD tree costs a
+    request. Parts no tree carries stay external.
+    """
+    resolved: dict[str, str | None] = {}
+    out = []
+    for display, ref, tokens, hashes, parts, citation in refs:
+        rewritten = []
+        for part in parts:
+            if is_external_citation(part.path):
+                if part.path not in resolved:
+                    resolved[part.path] = resolve_project_word(part.path, tree_order())
+                if resolved[part.path]:
+                    part = RefPart(resolved[part.path], part.start, part.end, part.raw)
+            rewritten.append(part)
+        out.append((display, ref, tokens, hashes, rewritten, citation))
+    return out
+
+
 def build_report(
     name: str, profile: dict, cache_dir: str, offline: bool = False
 ) -> ProfileReport:
@@ -1545,26 +1568,13 @@ def build_report(
                 candidates = near
         return None, candidates
 
-    def _tree_order():
-        for view in views:
-            yield _pin_tree_for(view)
-        for view in views:
-            yield _context_for(view)[1]
-
-    resolved_words: dict[str, str | None] = {}
-    for index, (display, ref, tokens, hashes, parts, citation) in enumerate(refs):
-        rewritten = []
-        for part in parts:
-            if is_external_citation(part.path):
-                if part.path not in resolved_words:
-                    resolved_words[part.path] = resolve_project_word(
-                        part.path, _tree_order()
-                    )
-                found = resolved_words[part.path]
-                if found:
-                    part = RefPart(found, part.start, part.end, part.raw)
-            rewritten.append(part)
-        refs[index] = (display, ref, tokens, hashes, rewritten, citation)
+    # Pin trees before HEAD trees, each read only when reached.
+    refs = _resolve_project_words(
+        refs,
+        lambda: itertools.chain(
+            map(_pin_tree_for, views), map(lambda v: _context_for(v)[1], views)
+        ),
+    )
 
     lines_cache: dict[tuple[str, str, int | None], list[str] | None] = {}
 
@@ -2571,6 +2581,25 @@ def _realign_part(
     return "skip", f"{part.path} absent at the writing revision"
 
 
+def _writing_pairs(
+    document: dict, repos: list, revisions: list[tuple[str, dict]], intro_sha: str
+) -> tuple[list, list[str]]:
+    """(repo, writing pin, current pin) per declared repository, and the
+    repositories whose writing pin the history cannot single out."""
+    pairs = []
+    ambiguous: list[str] = []
+    for pin_field, repo in repos:
+        current = document.get(f"{pin_field}_commit")
+        if not isinstance(current, str) or not current:
+            continue
+        written = _writing_pins(revisions, intro_sha, f"{pin_field}_commit")
+        if len(written) > 1:
+            ambiguous.append(f"{pin_field}_commit {' -> '.join(written)}")
+        elif written and written[0] != current:
+            pairs.append((repo, written[0], current))
+    return pairs, ambiguous
+
+
 def realign_prose(
     path: Path, cache_dir: str, offline: bool = False, dry_run: bool = False
 ) -> list[str]:
@@ -2629,17 +2658,7 @@ def realign_prose(
         if intro_sha is None:
             # The scalar is not committed yet: written now, under this pin.
             continue
-        pairs = []
-        ambiguous: list[str] = []
-        for pin_field, repo in repos:
-            current = document.get(f"{pin_field}_commit")
-            if not isinstance(current, str) or not current:
-                continue
-            written = _writing_pins(revisions, intro_sha, f"{pin_field}_commit")
-            if len(written) > 1:
-                ambiguous.append(f"{pin_field}_commit {' -> '.join(written)}")
-            elif written and written[0] != current:
-                pairs.append((repo, written[0], current))
+        pairs, ambiguous = _writing_pairs(document, repos, revisions, intro_sha)
         if ambiguous:
             # Moving from the wrong one rewrites a correct citation onto
             # someone else's code, with nothing to show it happened.
@@ -2985,9 +3004,8 @@ def _print_triage(args, selected: dict, reports: list[ProfileReport]) -> None:
         print(f"{score:5d}  {report.name:30s}  {state}")
 
 
-def main() -> None:
-    """Build one report per selected profile, then apply writes and output."""
-    args = build_parser().parse_args()
+def _refuse_flag_combinations(args) -> None:
+    """Exit on a flag the chosen mode would not apply, before any request."""
     if args.accept_changed and not args.emulator:
         print(
             "--accept-changed applies to one profile at a time: its diff has "
@@ -3052,6 +3070,12 @@ def main() -> None:
             file=sys.stderr,
         )
         raise SystemExit(1)
+
+
+def main() -> None:
+    """Build one report per selected profile, then apply writes and output."""
+    args = build_parser().parse_args()
+    _refuse_flag_combinations(args)
     profiles = load_emulator_profiles(args.emulators_dir, skip_aliases=False)
     selected = select_profiles(profiles, args)
     _check_quota(len(selected), args.offline)

scripts/validate_pr.py

@@ -22,7 +22,6 @@ import json
 import os
 import subprocess
 import sys
-from pathlib import Path
 
 sys.path.insert(0, os.path.dirname(__file__))
 from common import (