refactor: annotate the shared pack and lock api

scripts/artifacts.py

@@ -6,6 +6,7 @@ it is only meaningful if the comparison ignores the hour."""
 
 from __future__ import annotations
 
+from collections.abc import Callable, Iterator
 import contextlib
 import os
 import tempfile
@@ -22,7 +23,9 @@ _TIMESTAMP_PATTERNS = [
     re.compile(r'<div class="rb-timestamp">[^<]*</div>'),
 ]
 
-def write_if_changed(path: str, content: str, normalize=None) -> bool:
+def write_if_changed(
+    path: str, content: str, normalize: Callable[[str], str] | None = None
+) -> bool:
     """Write content to path only if the non-timestamp content differs.
 
     Compares new and existing content after stripping timestamp lines.
@@ -86,7 +89,7 @@ def write_bytes_atomic(path: str, content: bytes) -> None:
 
 
 @contextlib.contextmanager
-def file_lock(lock_path: str | os.PathLike, shared: bool = False):
+def file_lock(lock_path: str | os.PathLike, shared: bool = False) -> Iterator[None]:
     """Hold a lock on lock_path, waiting for it if taken.
 
     Several sessions refresh the same caches: without it, two swaps of one
@@ -140,7 +143,7 @@ HELD_LOCK_ENV = "RETROBIOS_HELD_ARTIFACT_DIR"
 
 
 @contextlib.contextmanager
-def artifact_lock(directory: str, exclusive: bool = True):
+def artifact_lock(directory: str, exclusive: bool = True) -> Iterator[None]:
     """Serialize access to a shared artifact directory across processes.
 
     Two pipeline runs building the same dist/ leave readers looking at
@@ -175,7 +178,7 @@ def artifact_lock(directory: str, exclusive: bool = True):
 
 
 @contextlib.contextmanager
-def hold_artifact_lock(directory: str):
+def hold_artifact_lock(directory: str) -> Iterator[None]:
     """Hold a directory exclusively across several child steps.
 
     A lock taken step by step frees the directory between them: another

scripts/common.py

@@ -6,6 +6,7 @@ and file resolution - eliminates DRY violations across scripts.
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from collections import Counter
 
 import hashlib
@@ -619,7 +620,7 @@ def declared_hash_verdict(
     md5_list: list,
     crc_candidates: set,
     zipped_file: str | None,
-    declared_size,
+    declared_size: int | list[int] | None,
     file_entry: dict | None = None,
     db: dict | None = None,
 ) -> str:
@@ -669,7 +670,11 @@ def declared_hash_verdict(
     return "data_dir_hash_exact"
 
 
-def _resolve_in_data_dirs(names_to_try, data_dir_registry, verdict):
+def _resolve_in_data_dirs(
+    names_to_try: list[str],
+    data_dir_registry: dict,
+    verdict: Callable[[str], str],
+) -> tuple[tuple[str, str] | None, str | None]:
     """Look for one of these names in the cached data directories.
 
     Returns the first candidate whose content satisfies the entry, and
@@ -743,7 +748,9 @@ def forget_data_dir_index() -> None:
     _data_dir_index.clear()
 
 
-def _resolve_agnostic(file_entry: dict, files_db: dict, has_strong_hash: bool):
+def _resolve_agnostic(
+    file_entry: dict, files_db: dict, has_strong_hash: bool
+) -> tuple[str, str] | None:
     """Any file of the right shape under the declared prefix.
 
     Some cores accept whatever filename sits in their BIOS directory, so the
@@ -2380,7 +2387,7 @@ from dumpcatalog import (  # noqa: E402,F401
 )
 
 
-def run_cli(main) -> None:
+def run_cli(main: Callable[[], int | None]) -> None:
     """Run a command-line entry point to its exit status.
 
     A reader that stops early (`verify.py --all | head`) closed the pipe

scripts/generate_pack.py

@@ -12,6 +12,8 @@ and 3-tier storage (embedded/external/user_provided).
 
 from __future__ import annotations
 
+from collections.abc import Callable, Iterator
+
 from artifacts import _build_timestamp
 import argparse
 import contextlib
@@ -331,7 +333,7 @@ def _data_directory_members(
     seen_destinations: set,
     seen_lower: set,
     seen_parents: set,
-):
+) -> Iterator[tuple[str, str]]:
     """Yield (source, destination) for the cached data directories a
     platform's systems declare.
 
@@ -386,7 +388,7 @@ def _data_directory_members(
 
 
 def _pack_data_directories(
-    zf,
+    zf: zipfile.ZipFile,
     pack_systems: dict,
     data_registry: dict | None,
     platform_name: str,
@@ -458,10 +460,10 @@ def _select_variants(
     emulators_dir: str,
     db: dict,
     base_dest: str,
-    emu_profiles,
-    target_cores,
+    emu_profiles: dict | None,
+    target_cores: set[str] | None,
     source: str,
-    regions,
+    regions: list[str] | None,
     one_per_slot: bool,
     required_only: bool = False,
 ) -> tuple[set, list, list]:
@@ -1129,7 +1131,7 @@ def generate_pack(
 
 def _extract_zip_to_archive(
     source_zip: str, dest_prefix: str, target_zf: zipfile.ZipFile
-):
+) -> None:
     """Extract contents of a source ZIP into target ZIP under dest_prefix."""
     with zipfile.ZipFile(source_zip, "r") as src:
         for info in src.infolist():
@@ -1153,7 +1155,7 @@ def _add_zip_to_pack(
     dest_path: str,
     target_zf: zipfile.ZipFile,
     file_entry: dict | None = None,
-):
+) -> None:
     """Write a source ZIP into the pack.
 
     A declared outer hash (md5/sha1/crc32) means the platform or profile
@@ -1179,7 +1181,7 @@ def _add_zip_to_pack(
 
 def _normalize_zip_for_pack(
     source_zip: str, dest_path: str, target_zf: zipfile.ZipFile
-):
+) -> None:
     """Add a MAME BIOS ZIP to the pack as a deterministic rebuild.
 
     Instead of copying the original ZIP (with non-deterministic metadata),
@@ -2151,7 +2153,7 @@ def generate_target_manifests(targets_dir: str, output_dir: str) -> None:
     print(f"Generated {count} target manifest(s) in {output_dir}")
 
 
-def _validate_args(args, parser):
+def _validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
     """Validate argument combinations and mutual exclusion rules."""
     has_platform = bool(args.platform)
     has_all = args.all
@@ -2251,8 +2253,13 @@ def _write_manifest_if_changed(path: str, manifest: dict) -> None:
 
 
 def _run_manifest_mode(
-    args, groups, db, zip_contents, emu_profiles, target_cores_cache
-):
+    args: argparse.Namespace,
+    groups: list[tuple[list[str], str]],
+    db: dict,
+    zip_contents: dict,
+    emu_profiles: dict,
+    target_cores_cache: dict[str, set[str] | None],
+) -> None:
     """Generate JSON manifests instead of ZIP packs."""
     registry_path = os.path.join(args.platforms_dir, "_registry.yml")
     os.makedirs(args.output_dir, exist_ok=True)
@@ -2350,7 +2357,7 @@ def _run_manifest_mode(
 
 
 @contextlib.contextmanager
-def _pack_output_lock(output_dir: str, exclusive: bool = True):
+def _pack_output_lock(output_dir: str, exclusive: bool = True) -> Iterator[None]:
     """Hold the output directory for the duration of a pack run."""
     try:
         with artifact_lock(output_dir, exclusive=exclusive):
@@ -2360,7 +2367,7 @@ def _pack_output_lock(output_dir: str, exclusive: bool = True):
         sys.exit(1)
 
 
-def _run_verify_packs(args):
+def _run_verify_packs(args: argparse.Namespace) -> None:
     """Verify each pack in the output directory against its platform.
 
     Delegates to verify_pack_against_platform so the command answers the same
@@ -2445,7 +2452,7 @@ def _pack_label(source: str, required_only: bool) -> str:
 
 
 def _truth_layer_is_empty(
-    platform: str, platforms_dir: str, emu_profiles: dict, target_cores
+    platform: str, platforms_dir: str, emu_profiles: dict, target_cores: set[str] | None
 ) -> bool:
     """Whether a platform runs no core a profile describes.
 
@@ -2456,8 +2463,14 @@ def _truth_layer_is_empty(
     return not _platform_cores(config, emu_profiles or {}, target_cores=target_cores)
 
 
-def _skip_empty_truth(args, source: str, representative: str, emu_profiles, tc,
-                      failed: list[str]) -> bool:
+def _skip_empty_truth(
+    args: argparse.Namespace,
+    source: str,
+    representative: str,
+    emu_profiles: dict,
+    tc: set[str] | None,
+    failed: list[str],
+) -> bool:
     """Leave out a truth pack that would be empty; refuse it when asked for."""
     if source != "truth" or not _truth_layer_is_empty(
         representative, args.platforms_dir, emu_profiles, tc
@@ -2475,21 +2488,21 @@ def _skip_empty_truth(args, source: str, representative: str, emu_profiles, tc,
     return True
 
 
-def _produced(split: bool, zip_path, zip_paths) -> bool:
+def _produced(split: bool, zip_path: str | None, zip_paths: list[str] | None) -> bool:
     """Whether the build wrote what the run asked for."""
     return bool(zip_paths if split else zip_path)
 
 
 def _run_platform_packs(
-    args,
-    groups,
-    db,
-    zip_contents,
-    data_registry,
-    emu_profiles,
-    target_cores_cache,
-    system_filter,
-):
+    args: argparse.Namespace,
+    groups: list[tuple[list[str], str]],
+    db: dict,
+    zip_contents: dict,
+    data_registry: dict | None,
+    emu_profiles: dict,
+    target_cores_cache: dict[str, set[str] | None],
+    system_filter: list[str] | None,
+) -> None:
     """Generate ZIP packs for platform groups and verify."""
     if args.all_variants:
         variants = [
@@ -2632,7 +2645,7 @@ def _run_platform_packs(
         sys.exit(1)
 
 
-def _refuse_for_all_variants(args, parser) -> None:
+def _refuse_for_all_variants(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
     """Flags --all-variants would override, or modes that never read it."""
     if not args.all_variants:
         return
@@ -2662,7 +2675,7 @@ def _refuse_for_all_variants(args, parser) -> None:
                 parser.error(f"{flag} is incompatible with --verify-packs")
 
 
-def _refuse_refresh_data(args, parser) -> None:
+def _refuse_refresh_data(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
     """Only the platform pack path refreshes the data directories.
 
     Elsewhere the cache on disk was packed as it stood, and --offline wins
@@ -2684,14 +2697,16 @@ def _refuse_refresh_data(args, parser) -> None:
                 parser.error(f"{name} is incompatible with {flag}")
 
 
-def _refuse_include_archived(args, parser) -> None:
+def _refuse_include_archived(
+    args: argparse.Namespace, parser: argparse.ArgumentParser
+) -> None:
     """Only --all chooses among registered platforms; every other mode names
     its own platform, emulator, system or hash."""
     if args.include_archived and not args.all:
         parser.error("--include-archived requires --all")
 
 
-def _refuse_unapplied_flags(args, parser) -> None:
+def _refuse_unapplied_flags(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
     """Refuse every flag the requested mode would not apply.
 
     A mode applies a narrowing flag or refuses it: a swallowed flag answers
@@ -2749,7 +2764,7 @@ QUICK_MODE_FLAGS: dict[str, set[str]] = {
 }
 
 
-def _given_flags(args) -> set[str]:
+def _given_flags(args: argparse.Namespace) -> set[str]:
     """Flags that select or narrow what a run produces, as typed."""
     flags = {
         "--platform": args.platform,
@@ -2787,7 +2802,7 @@ def _given_flags(args) -> set[str]:
 _MISSING_DATA_DIRS: set[str] = set()
 
 
-def main():
+def main() -> None:
     with contextlib.ExitStack() as held:
         _main(held)
     if _MISSING_DATA_DIRS:
@@ -3281,8 +3296,8 @@ def _attach_claims(
     claims: list[dict],
     db: dict,
     bios_dir: str,
-    zip_contents,
-    offline,
+    zip_contents: dict | None,
+    offline: bool | None,
     required_only: bool,
     validation_index: dict | None,
     owners: dict | None,
@@ -3353,7 +3368,7 @@ def _record_slot_choices(
     entry: dict,
     choices: list[tuple[str, str | None]],
     declarations: list[tuple[list[str], dict, str]],
-    settle,
+    settle: Callable[[dict, str, str, str | None], tuple[str | None, str]],
     dest: str,
     db: dict,
     repo_root: str,
@@ -3416,8 +3431,8 @@ def _manifest_core_entries(
     bios_dir: str,
     base_dest: str,
     repo_root: str,
-    zip_contents,
-    offline,
+    zip_contents: dict | None,
+    offline: bool | None,
     region_drops: set,
     case_insensitive: bool,
     seen_destinations: set,
@@ -3425,7 +3440,7 @@ def _manifest_core_entries(
     seen_parents: set,
     manifest_files: list,
     omitted_by_destination: dict,
-    record_omission,
+    record_omission: Callable[..., None],
     pack_only_sizes: list[int],
     selection: SelectionIndex,
     required_only: bool = False,
@@ -3612,7 +3627,7 @@ def _record_absent(
     required_only: bool,
     seen_destinations: set,
     omitted_by_destination: dict,
-    record_omission,
+    record_omission: Callable[..., None],
 ) -> None:
     """A file a core needs that the collection does not hold is an omission.
 
@@ -3641,7 +3656,7 @@ def _record_absent(
         )
 
 
-def _record_unplaceable(unplaceable: list[dict], record_omission) -> None:
+def _record_unplaceable(unplaceable: list[dict], record_omission: Callable[..., None]) -> None:
     """Held by the collection, nowhere to put it on this platform.
 
     Said in the manifest, where RomM's DOSBox Pure ROMs were simply absent.
@@ -3831,7 +3846,9 @@ def generate_manifest(
                     del remaining[sid]
             return order
 
-        def settle(file_entry: dict, sys_id: str, dest: str, override: str | None):
+        def settle(
+            file_entry: dict, sys_id: str, dest: str, override: str | None
+        ) -> tuple[str | None, str]:
             """The file served once the slot is settled, or None and why not."""
             local_path, status = slots.apply_override(override, *resolve_file(
                 file_entry,
@@ -4056,7 +4073,7 @@ def _manifest_core_index(
     config: dict,
     profiles: dict[str, dict],
     files: list[dict],
-    omitted,
+    omitted: list[dict],
 ) -> tuple[dict[str, list[str]], dict[str, list[str]]]:
     """What the installer needs to narrow by a list of core names.
 

scripts/validation.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 
 import os
 import zipfile
-from collections.abc import Mapping
+from collections.abc import Iterator, Mapping
 
 from common import (
     adler32_of,
@@ -574,7 +574,7 @@ def find_validated_variant(
     current_real = os.path.realpath(current_path)
     seen: set[str] = set()
 
-    def candidates():
+    def candidates() -> Iterator[str]:
         # Sorted: the index holds sets, whose order follows the per-process
         # string hash, and the first passing candidate is the one shipped.
         expected = validation_index[fname]
@@ -789,10 +789,10 @@ def validated_choice(
         # The cores that read this destination judge it, each by its own
         # entries.
         judges = owning & set(rules["emulators"])
-    def check(path: str, strict: bool):
+    def check(path: str, strict: bool) -> tuple[str, list[str]] | None:
         return check_file_validation(path, name, validation_index, bios_dir, judges, strict)
 
-    def variant(strict: bool):
+    def variant(strict: bool) -> str | None:
         return find_validated_variant(
             file_entry, db, local_path, validation_index, bios_dir,
             platform_digest=platform_digest, emulators=judges, strict=strict,

tests/test_audit_regressions.py

@@ -3747,5 +3747,38 @@ class TargetListsReadThePinnedRelease(unittest.TestCase):
                 self.assertNotIn("fetch_github_latest_version(", source)
 
 
+
+class SignaturesCarryTheirTypes(unittest.TestCase):
+    """Type hints on every signature is a written rule; 177 had none, the
+    shared API among them (artifact locks, the data-dir resolver, the region
+    priority list of _select_variants)."""
+
+    # Lowered as modules are annotated; never raised.
+    CEILING = 139
+    COMPLETE = ("artifacts.py", "common.py", "validation.py", "generate_pack.py")
+
+    @staticmethod
+    def _missing(path: Path) -> list[str]:
+        found = []
+        for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))):
+            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                continue
+            args = node.args
+            params = [a for a in (*args.posonlyargs, *args.args, *args.kwonlyargs)
+                      if a.arg not in ("self", "cls")]
+            params += [a for a in (args.vararg, args.kwarg) if a]
+            if node.returns is None or any(a.annotation is None for a in params):
+                found.append(f"{path.name}:{node.lineno} {node.name}")
+        return found
+
+    def test_the_count_only_goes_down(self):
+        paths = [ROOT / "install.py", *sorted((ROOT / "scripts").rglob("*.py"))]
+        missing = [m for path in paths for m in self._missing(path)]
+        self.assertLessEqual(len(missing), self.CEILING, missing[:20])
+        for name in self.COMPLETE:
+            with self.subTest(module=name):
+                self.assertEqual(self._missing(ROOT / "scripts" / name), [])
+
+
 if __name__ == "__main__":
     unittest.main()