Added umu download and repair for missing components on install, update and manual repair.

2026-09-02 21:42:49 +02:00
parent 6f20b5e359
commit 7ce6ac3dd7
9 changed files with 660 additions and 40 deletions
--- a/ptlinux/main.py
+++ b/ptlinux/main.py
@@ -9,10 +9,16 @@ def main() -> int:
 	arguments = sys.argv[1:]
 	if not arguments or arguments == ["gui"]:
 		from .paths import AppPaths
-		from .ui import run_dashboard
+		from .umu import provision_umu_for_startup

 		try:
-			run_dashboard(AppPaths.from_environment())
+			paths = AppPaths.from_environment()
+			# Older installed releases cannot run a bootstrap step that only exists
+			# in their update payload. Their automatic restart enters here instead.
+			provision_umu_for_startup(paths)
+			from .ui import run_dashboard
+
+			run_dashboard(paths)
 		except KeyboardInterrupt:
 			return 130
 		except (ValueError, OSError, RuntimeError) as exc:
--- a/ptlinux/cli.py
+++ b/ptlinux/cli.py
@@ -4,10 +4,11 @@ from __future__ import annotations

 import argparse
 from contextlib import AbstractContextManager, nullcontext
-from dataclasses import asdict
+from dataclasses import asdict, replace
 from datetime import datetime
 import json
 from pathlib import Path
+import re
 import sys
 import time
 from typing import Callable, Sequence
@@ -20,7 +21,7 @@ from ptdiag.logs import (
 )

 from . import __version__
-from .command import CommandError, CommandRunner
+from .command import CommandError, CommandOutcome, CommandRunner
 from .depot import (
 	SyncProgress,
 	read_cached_depot_launch_catalog,
@@ -44,6 +45,7 @@ from .login_trace import DEFAULT_TRACE_TIMEOUT_SECONDS, LOGIN_TRACE_DISABLED_REA
 from .overlay import cache_compatibility_binaries, prepare_runtime_view
 from .paths import AppPaths
 from .runtime import (
+	RuntimeErrorPT,
 	game_runtime_ready,
 	prepare_game_runtime,
 	run_game,
@@ -57,6 +59,12 @@ from .self_update import (
 	update_check_to_dict,
 	verify_launcher_update_channel,
 )
+from .umu import (
+	UmuBootstrapResult,
+	find_umu_launcher,
+	provision_umu,
+	provision_umu_for_startup,
+)


 def build_parser() -> argparse.ArgumentParser:
@@ -74,6 +82,10 @@ def build_parser() -> argparse.ArgumentParser:

 	subparsers.add_parser("gui", help="open the native Linux dashboard")
 	subparsers.add_parser("doctor", help="check cross-distribution dependencies")
+	subparsers.add_parser(
+		"runtime-bootstrap",
+		help="download and verify the Steam Runtime launcher helper",
+	)
 	login_diagnostics = subparsers.add_parser(
 		"login-diagnostics",
 		help="check the client, runtime, and publisher login route without credentials",
@@ -236,12 +248,15 @@ def main(argv: Sequence[str] | None = None) -> int:
 	paths = AppPaths.from_environment()
 	try:
 		if command == "gui":
+			provision_umu_for_startup(paths)
 			from .ui import run_dashboard

 			run_dashboard(paths)
 			return 0
 		if command == "doctor":
 			return _doctor(arguments.json)
+		if command == "runtime-bootstrap":
+			return _runtime_bootstrap(paths, arguments.json)
 		if command == "login-diagnostics":
 			return _login_diagnostics(
 				paths,
@@ -368,6 +383,25 @@ def _doctor(as_json: bool) -> int:
 	return 0 if report.ok else 1


+def _runtime_bootstrap(paths: AppPaths, as_json: bool) -> int:
+	_print_operation_status(as_json, 1, 1, "Preparing the Steam Runtime helper…")
+	result = provision_umu(paths)
+	executable = find_umu_launcher(paths)
+	if executable is None:
+		raise LauncherError("the Steam Runtime helper is still unavailable")
+	_emit(
+		{
+			"ok": True,
+			"runtime_bootstrap": {
+				"executable": executable,
+				"managed_download": asdict(result) if result else None,
+			},
+		},
+		as_json,
+	)
+	return 0
+
+
 def _login_diagnostics(
 	paths: AppPaths,
 	timeout_seconds: float,
@@ -781,8 +815,13 @@ def _play_locked(
 			)
 	elif not dry_run:
 		overlay = prepare_runtime_view(paths, source)
-	_print_operation_status(as_json, 2, 4, "Selecting the configured game runtime…")
-	selection = select_runtime(settings)
+	_print_operation_status(as_json, 2, 4, "Ensuring the compatibility runtime is available…")
+	umu_bootstrap = (
+		None
+		if reuse_prepared
+		else _ensure_portable_umu_for_play(paths, settings, dry_run=dry_run)
+	)
+	selection = select_runtime(settings, paths=paths)
 	runner = _runner(paths, "play", dry_run)
 	setup_results = []
 	if reuse_prepared:
@@ -820,6 +859,7 @@ def _play_locked(
 			start_option=start_option,
 			use_gamebox=not no_box,
 			concurrent_prefix_launch=reuse_prepared,
+			check=False,
 		)
 	except KeyboardInterrupt:
 		current_game_log = _current_game_log(
@@ -841,23 +881,63 @@ def _play_locked(
 		and runtime_issues["host_futex_fatal_count"]
 	):
 		runtime_issues["recommended_one_variable_comparison"] = "proton7"
-	_emit(
-		{
-			"ok": outcome.returncode == 0,
-			"dry_run": dry_run,
-			"runtime": asdict(selection),
-			"realm": asdict(start_option),
-			"runtime_view": asdict(overlay) if overlay else None,
-			"setup_commands": [asdict(result) for result in setup_results],
-			"game": asdict(outcome),
-			"current_game_log": current_game_log,
-			"runtime_issues": runtime_issues,
-		},
-		as_json,
-	)
+	elif selection.kind == "proton7" and runtime_issues["gstreamer_missing_libraries"]:
+		runtime_issues["recommended_one_variable_comparison"] = "proton7-container"
+	elif runtime_issues["dbghelp_symdeferred_assertion_count"]:
+		runtime_issues["recommended_one_variable_comparison"] = "umu"
+	report = {
+		"ok": outcome.returncode == 0,
+		"dry_run": dry_run,
+		"runtime": asdict(selection),
+		"umu_bootstrap": asdict(umu_bootstrap) if umu_bootstrap else None,
+		"realm": asdict(start_option),
+		"runtime_view": asdict(overlay) if overlay else None,
+		"setup_commands": [asdict(result) for result in setup_results],
+		"game": asdict(outcome),
+		"current_game_log": current_game_log,
+		"runtime_issues": runtime_issues,
+	}
+	if as_json or dry_run:
+		_emit(report, as_json)
+	else:
+		_print_game_outcome(outcome, current_game_log, runtime_issues)
 	return 0 if outcome.returncode == 0 else 1


+def _ensure_portable_umu_for_play(
+	paths: AppPaths,
+	settings: Settings,
+	*,
+	dry_run: bool,
+) -> UmuBootstrapResult | None:
+	"""Provide a pinned UMU when the selected policy may use a container."""
+
+	if settings.runtime == "proton7":
+		return None
+	if dry_run:
+		return None
+	try:
+		return provision_umu(paths)
+	except (OSError, RuntimeError) as exc:
+		if settings.runtime != "auto":
+			raise LauncherError(
+				f"the verified portable UMU runtime helper could not be installed: {exc}"
+			) from exc
+		try:
+			select_runtime(replace(settings, runtime="proton7"), paths=paths)
+		except RuntimeErrorPT:
+			raise LauncherError(
+				"no usable game runtime is available and the verified portable UMU "
+				f"runtime helper could not be installed: {exc}"
+			) from exc
+		print(
+			"Warning: the Steam Runtime helper could not be installed; falling back "
+			f"to direct Proton 7 for this launch ({exc}).",
+			file=sys.stderr,
+		)
+		return None
+
+
 def _current_game_log(
 	directory: Path,
 	baseline: dict[str, LogFileState],
@@ -910,15 +990,76 @@ def _runtime_issue_summary(stderr: str) -> dict[str, object]:
 	folded = stderr.casefold()
 	futex_message = "the futex facility returned an unexpected error code"
 	wrong_elf_message = "wrong elf class: elfclass64"
+	dbghelp_assertion = "assertion failed: module->module.symtype != symdeferred"
+	missing_libraries = sorted(
+		{
+			library
+			for line in folded.splitlines()
+			if "gstreamer-warning" in line and "failed to load plugin" in line
+			for library in re.findall(
+				r"([a-z0-9_.+-]+\.so(?:\.\d+)*): cannot open shared object file",
+				line,
+			)
+		}
+	)
 	return {
 		"host_futex_fatal_count": folded.count(futex_message),
 		"gstreamer_wrong_elf_class_count": folded.count(wrong_elf_message),
+		"gstreamer_missing_library_count": len(missing_libraries),
+		"gstreamer_missing_libraries": missing_libraries,
 		"gstreamer_stream_error": "winegstreamer error" in folded,
 		"mesa_egl_initialization_failed": "egl: failed to create dri2 screen" in folded,
 		"steam_api_app_id_missing": "steamapi_init() failed; no appid found" in folded,
+		"dbghelp_symdeferred_assertion_count": folded.count(dbghelp_assertion),
 	}


+def _print_game_outcome(
+	outcome: CommandOutcome,
+	current_game_log: dict[str, object] | None,
+	runtime_issues: dict[str, object],
+) -> None:
+	# Keep the human-facing stream concise. CommandRunner already relayed every
+	# process line live; replaying its capture here made each Proton failure appear
+	# twice in the dashboard.
+	if outcome.timed_out:
+		print("Game client timed out and was stopped.", file=sys.stderr)
+	elif outcome.returncode == 0:
+		print("Game client exited normally.")
+	else:
+		print(f"Game client exited with status {outcome.returncode}.", file=sys.stderr)
+	if current_game_log is not None:
+		_print_interrupted_game_log(current_game_log)
+	wrong_elf_count = runtime_issues["gstreamer_wrong_elf_class_count"]
+	if isinstance(wrong_elf_count, int) and wrong_elf_count:
+		print(
+			"The repeated GStreamer wrong-ELF-class messages are plugin-scan noise; "
+			"they do not mean that Proton itself failed to start.",
+			file=sys.stderr,
+		)
+	missing_libraries = runtime_issues["gstreamer_missing_libraries"]
+	if isinstance(missing_libraries, list) and missing_libraries:
+		print(
+			"Media plugins could not load these shared libraries: "
+			+ ", ".join(str(value) for value in missing_libraries)
+			+ ".",
+			file=sys.stderr,
+		)
+	assertion_count = runtime_issues["dbghelp_symdeferred_assertion_count"]
+	if isinstance(assertion_count, int) and assertion_count:
+		print(
+			"Proton 7's older Wine build hit its dbghelp SymDeferred assertion; this "
+			"can mask the client failure that occurred immediately before it.",
+			file=sys.stderr,
+		)
+	recommendation = runtime_issues.get("recommended_one_variable_comparison")
+	if isinstance(recommendation, str):
+		print(
+			f"Recommended single-variable runtime comparison: {recommendation}.",
+			file=sys.stderr,
+		)
+
+
 def _print_interrupted_game_log(report: dict[str, object]) -> None:
 	print(f"\nCurrent game log: {report['path']}", file=sys.stderr)
 	if report["module_code_sync_error"]:
@@ -1020,17 +1161,44 @@ def _launcher_update(
 			as_json,
 		)
 		return 0
-	_print_operation_status(as_json, 1, 2, "Verifying the signed launcher release…")
+	_print_operation_status(as_json, 1, 3, "Verifying the signed launcher release…")
 	result = install_launcher_update(
 		paths,
 		manifest_url=manifest_url,
 		public_key_path=public_key,
 	)
-	_print_operation_status(as_json, 2, 2, "Launcher update installed safely.")
+	_print_operation_status(
+		as_json,
+		2,
+		3,
+		"Launcher update installed; preparing the Steam Runtime helper…",
+	)
+	umu_result: UmuBootstrapResult | None = None
+	umu_executable: str | None = None
+	umu_warning: str | None = None
+	try:
+		umu_result = provision_umu(paths)
+		umu_executable = find_umu_launcher(paths)
+	except (OSError, RuntimeError) as exc:
+		# Activation is already durable. Keep the update successful so the GUI can
+		# restart into the new release, whose startup hook and Play path both retry.
+		warning_message = (
+			"the launcher update was installed, but the Steam Runtime helper could "
+			f"not be prepared yet: {exc}"
+		)
+		umu_warning = warning_message
+		if not as_json:
+			print(f"Warning: {warning_message}", file=sys.stderr)
+	_print_operation_status(as_json, 3, 3, "Launcher update and runtime check complete.")
 	_emit(
 		{
 			"ok": True,
 			"launcher_update": asdict(result),
+			"runtime_bootstrap": {
+				"executable": umu_executable,
+				"managed_download": asdict(umu_result) if umu_result else None,
+				"warning": umu_warning,
+			},
 		},
 		as_json,
 	)
--- a/ptlinux/dashboard_debug.py
+++ b/ptlinux/dashboard_debug.py
@@ -65,7 +65,7 @@ def inspect_runtime_debug(
 	paths: AppPaths,
 	settings: Settings,
 	*,
-	selector: RuntimeSelector = select_runtime,
+	selector: RuntimeSelector | None = None,
 	sync_resolver: SyncResolver = resolve_proton_sync_mode,
 ) -> RuntimeDebugInfo:
 	"""Resolve the next runtime and compare it with the preparation marker."""
@@ -85,7 +85,11 @@ def inspect_runtime_debug(
 	selection: RuntimeSelection | None = None
 	error: str | None = None
 	try:
-		selection = selector(settings)
+		selection = (
+			select_runtime(settings, paths=paths)
+			if selector is None
+			else selector(settings)
+		)
 	except (OSError, RuntimeErrorPT, ValueError) as exc:
 		error = str(exc).strip() or type(exc).__name__
 	selected_kind = selection.kind if selection is not None else None
--- a/ptlinux/doctor.py
+++ b/ptlinux/doctor.py
@@ -11,7 +11,9 @@ import shutil
 from typing import Mapping

 from .launcher import find_wine_tools
+from .paths import AppPaths
 from .runtime import find_steam_proton7
+from .umu import managed_umu_ready


 ARCH_FAMILY_DISTRIBUTIONS = frozenset(
@@ -48,6 +50,7 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
 	distribution = os_release.get("ID", "unknown")
 	version = os_release.get("VERSION_ID", "unknown")
 	wine_tools = find_wine_tools()
+	umu_path = _umu_path(env)
 	tools = [
 		ToolStatus(
 			"python3",
@@ -93,9 +96,9 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
 		),
 		ToolStatus(
 			"umu-run",
-			shutil.which("umu-run"),
+			umu_path,
 			distribution in ARCH_FAMILY_DISTRIBUTIONS,
-			"current portable Proton runtime; automatic on Arch-family hosts",
+			"Steam Runtime container; managed automatically when needed",
 		),
 		ToolStatus(
 			"vulkaninfo",
@@ -110,14 +113,15 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
 		warnings.append("PristonTale EU requires an x86-64 host with 32-bit Vulkan support.")
 	if not env.get("WAYLAND_DISPLAY") and not env.get("DISPLAY"):
 		warnings.append("No graphical X11 or Wayland session is visible.")
-	if shutil.which("umu-run") is None and not _has_steam_proton7(env):
+	if umu_path is None and not _has_steam_proton7(env):
 		warnings.append(
-			"No game runtime was found: install umu-launcher or Steam Proton 7.0."
+			"No local game runtime is installed yet; run runtime-bootstrap to fetch the "
+			"pinned official UMU helper. Startup and Play also retry automatically."
 		)
 	warnings.extend(_arch_runtime_warnings(distribution, platform.machine()))
 	required_ok = all(item.path is not None for item in tools if item.required)
 	return DoctorReport(
-		ok=required_ok and not any("No game runtime" in warning for warning in warnings),
+		ok=required_ok,
 		distribution=distribution,
 		version=version,
 		architecture=platform.machine(),
@@ -180,6 +184,19 @@ def _has_steam_proton7(environment: Mapping[str, str]) -> bool:
 	return find_steam_proton7(environment) is not None


+def _umu_path(environment: Mapping[str, str]) -> str | None:
+	system = shutil.which("umu-run", path=environment.get("PATH"))
+	if system:
+		return system
+	try:
+		paths = AppPaths.from_environment(environment)
+	except RuntimeError:
+		return None
+	if managed_umu_ready(paths.managed_umu_executable):
+		return str(paths.managed_umu_executable.resolve())
+	return None
+
+
 def _platform_optional_tools(distribution: str, *, architecture: str) -> list[ToolStatus]:
 	if architecture.casefold() not in {"x86_64", "amd64"}:
 		return []
--- a/ptlinux/login_diagnostics.py
+++ b/ptlinux/login_diagnostics.py
@@ -213,7 +213,7 @@ def inspect_login_environment(
 	)
 	runtime_requested = runtime_override or settings.runtime
 	diagnostic_settings = replace(settings, runtime=runtime_requested)
-	selection, runtime_error = _runtime_selection(diagnostic_settings)
+	selection, runtime_error = _runtime_selection(paths, diagnostic_settings)
 	distribution_id = detect_distribution_id() or "unknown"
 	kernel_release = platform.release()
 	proton_sync_effective = (
@@ -476,9 +476,12 @@ def _active_game_root(paths: AppPaths) -> Path:
 	return paths.source_game


-def _runtime_selection(settings: Settings) -> tuple[RuntimeSelection | None, str | None]:
+def _runtime_selection(
+	paths: AppPaths,
+	settings: Settings,
+) -> tuple[RuntimeSelection | None, str | None]:
 	try:
-		return select_runtime(settings), None
+		return select_runtime(settings, paths=paths), None
 	except RuntimeError as exc:
 		return None, str(exc)

--- a/ptlinux/paths.py
+++ b/ptlinux/paths.py
@@ -71,6 +71,23 @@ class AppPaths:
 	def downloads(self) -> Path:
 		return self.cache / "downloads"

+	@property
+	def runtime_tools(self) -> Path:
+		"""Launcher-managed helpers that are independent of app releases."""
+
+		return self.data / "runtime-tools"
+
+	@property
+	def managed_umu_executable(self) -> Path:
+		return self.runtime_tools / "bin/umu-run"
+
+	@property
+	def managed_umu_metadata(self) -> Path:
+		return self.state / "umu-launcher.json"
+
+	def managed_umu_archive(self, version: str) -> Path:
+		return self.downloads / f"umu-launcher-{version}-zipapp.tar"
+
 	@property
 	def app_home(self) -> Path:
 		"""Stable home containing versioned native-launcher releases."""
--- a/ptlinux/runtime.py
+++ b/ptlinux/runtime.py
@@ -20,6 +20,7 @@ from .depot import DEFAULT_DEPOT_START_OPTION, DepotStartOption
 from .gamebox import build_gamebox_command
 from .paths import AppPaths
 from .settings import Settings
+from .umu import find_umu_launcher


 WINDOWS_GAME_DIRECTORY = r"C:\Program Files (x86)\Pristontale EU"
@@ -61,11 +62,12 @@ def select_runtime(
 	settings: Settings,
 	environment: Mapping[str, str] | None = None,
 	*,
+	paths: AppPaths | None = None,
 	distribution_id: str | None = None,
 	kernel_release: str | None = None,
 ) -> RuntimeSelection:
 	env = os.environ if environment is None else environment
-	umu = shutil.which("umu-run", path=env.get("PATH"))
+	umu = find_umu_launcher(paths, env)
 	current_umu_host = prefers_current_umu_host(
 		distribution_id=distribution_id,
 		kernel_release=kernel_release,
@@ -257,7 +259,7 @@ def prepare_game_runtime(
 		paths.create_base_directories()
 	results: list[CommandOutcome] = []
 	if selection.kind == "umu":
-		environment = _umu_environment(paths)
+		environment = _umu_environment(paths, selection)
 		if not native_d3d_ready(paths.umu_prefix):
 			results.append(
 				runner.run(
@@ -398,7 +400,7 @@ def run_game(
 			native_executable,
 			*game_arguments,
 		]
-		environment = _umu_environment(paths)
+		environment = _umu_environment(paths, selection)
 	elif selection.kind == "proton7-container":
 		if not selection.proton_path:
 			raise RuntimeErrorPT("Proton script path is missing")
@@ -550,7 +552,7 @@ def run_runtime_prefix_command(
 		raise ValueError("runtime-prefix command arguments cannot be empty")
 	if selection.kind == "umu":
 		command = [selection.executable, *arguments]
-		environment = _umu_environment(paths)
+		environment = _umu_environment(paths, selection)
 	elif selection.kind == "proton7-container":
 		if not selection.proton_path or not selection.steam_root:
 			raise RuntimeErrorPT("incomplete Proton 7 runtime selection")
@@ -586,14 +588,19 @@ def _runtime_python_path(existing: str | None) -> str:
 	return os.pathsep.join((package_root, existing))


-def _umu_environment(paths: AppPaths) -> dict[str, str]:
-	return {
+def _umu_environment(
+	paths: AppPaths,
+	selection: RuntimeSelection,
+) -> dict[str, str]:
+	environment = {
 		"WINEPREFIX": str(paths.umu_prefix),
 		"GAMEID": "umu-pristontale-eu",
 		"STORE": "none",
 		"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
 		"WINEDEBUG": "-all",
 	}
+	environment.update(_portable_umu_environment(paths, selection))
+	return environment


 def _umu_proton7_environment(
@@ -619,10 +626,26 @@ def _umu_proton7_environment(
 		"DXVK_STATE_CACHE_PATH": str(paths.dxvk_cache),
 		"WINEDEBUG": "-all",
 	}
+	environment.update(_portable_umu_environment(paths, selection))
 	environment.update(_proton_sync_environment(synchronization))
 	return environment


+def _portable_umu_environment(
+	paths: AppPaths,
+	selection: RuntimeSelection,
+) -> dict[str, str]:
+	"""Keep downloads made by our zipapp inside the selected installation."""
+
+	try:
+		selected = Path(selection.executable).expanduser().resolve()
+	except (OSError, RuntimeError):
+		return {}
+	if selected != paths.managed_umu_executable.resolve():
+		return {}
+	return {"UMU_FOLDERS_PATH": str(paths.runtime_tools)}
+
+
 def _proton_environment(
 	paths: AppPaths,
 	selection: RuntimeSelection,
--- a/ptlinux/umu.py
+++ b/ptlinux/umu.py
@@ -0,0 +1,356 @@
+"""Pinned, user-level bootstrap for the official UMU launcher zipapp."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from datetime import datetime
+import hashlib
+import json
+import os
+from pathlib import Path
+import secrets
+import shutil
+import sys
+import tarfile
+import threading
+from typing import Callable, Mapping, Protocol
+from urllib.error import HTTPError
+from urllib.request import Request, urlopen
+import zipfile
+
+from . import __version__
+from .paths import AppPaths
+
+
+UMU_LAUNCHER_VERSION = "1.4.4"
+UMU_LAUNCHER_URL = (
+	"https://github.com/Open-Wine-Components/umu-launcher/releases/download/"
+	f"{UMU_LAUNCHER_VERSION}/umu-launcher-{UMU_LAUNCHER_VERSION}-zipapp.tar"
+)
+UMU_ARCHIVE_SHA256 = "eb590691841f7fad3fc3ad8fd5db4ccb87849fe7948e62b28ece7a4ee48cc851"
+UMU_EXECUTABLE_SHA256 = "d0005a58602041229cc467dab03dc0c0b9e8cce09a8145b16b7683244cf17804"
+UMU_ARCHIVE_MEMBER = "umu/umu-run"
+MAX_UMU_ARCHIVE_BYTES = 4 * 1024 * 1024
+MAX_UMU_EXECUTABLE_BYTES = 2 * 1024 * 1024
+READ_SIZE = 256 * 1024
+USER_AGENT = f"PT-Linux/{__version__} (+https://pristontale.eu/)"
+
+
+class UmuBootstrapError(RuntimeError):
+	"""Raised when the pinned portable UMU launcher cannot be prepared."""
+
+
+class _HttpResponse(Protocol):
+	headers: Mapping[str, str]
+
+	def read(self, size: int = -1) -> bytes:
+		...
+
+	def __enter__(self) -> _HttpResponse:
+		...
+
+	def __exit__(
+		self,
+		exception_type: object,
+		exception: object,
+		traceback: object,
+	) -> object:
+		...
+
+
+UmuOpener = Callable[..., _HttpResponse]
+
+
+@dataclass(frozen=True)
+class UmuBootstrapResult:
+	path: str
+	version: str
+	source_url: str
+	archive_sha256: str
+	executable_sha256: str
+	changed: bool
+
+
+def find_umu_launcher(
+	paths: AppPaths | None = None,
+	environment: Mapping[str, str] | None = None,
+) -> str | None:
+	"""Find a system UMU first, then the launcher's verified portable copy."""
+
+	env = os.environ if environment is None else environment
+	located = shutil.which("umu-run", path=env.get("PATH"))
+	if located:
+		candidate = Path(located).expanduser().resolve()
+		if (
+			paths is None
+			or candidate != paths.managed_umu_executable.resolve()
+			or managed_umu_ready(candidate)
+		):
+			return str(candidate)
+	if paths is not None and managed_umu_ready(paths.managed_umu_executable):
+		return str(paths.managed_umu_executable.resolve())
+	return None
+
+
+def managed_umu_ready(path: Path) -> bool:
+	"""Accept only the exact launcher zipapp pinned by this release."""
+
+	try:
+		return (
+			path.is_file()
+			and not path.is_symlink()
+			and os.access(path, os.X_OK)
+			and _sha256(path) == UMU_EXECUTABLE_SHA256
+		)
+	except OSError:
+		return False
+
+
+def provision_umu(
+	paths: AppPaths,
+	*,
+	environment: Mapping[str, str] | None = None,
+	timeout_seconds: float = 60.0,
+	opener: UmuOpener = urlopen,
+) -> UmuBootstrapResult | None:
+	"""Ensure that either a system or launcher-managed UMU is available."""
+
+	if find_umu_launcher(paths, environment) is not None:
+		return None
+	return ensure_managed_umu(
+		paths,
+		timeout_seconds=timeout_seconds,
+		opener=opener,
+	)
+
+
+def provision_umu_for_startup(
+	paths: AppPaths,
+	*,
+	timeout_seconds: float = 15.0,
+) -> UmuBootstrapResult | None:
+	"""Repair a missing helper without preventing dashboard startup on failure."""
+
+	try:
+		return provision_umu(paths, timeout_seconds=timeout_seconds)
+	except (OSError, RuntimeError) as exc:
+		print(
+			"Warning: the Steam Runtime helper could not be prepared during launcher "
+			f"startup; Play will retry ({exc}).",
+			file=sys.stderr,
+		)
+		return None
+
+
+def ensure_managed_umu(
+	paths: AppPaths,
+	*,
+	timeout_seconds: float = 60.0,
+	opener: UmuOpener = urlopen,
+) -> UmuBootstrapResult:
+	"""Install the pinned official zipapp without modifying the host Python."""
+
+	destination = paths.managed_umu_executable
+	if managed_umu_ready(destination):
+		return _result(destination, changed=False)
+
+	archive = paths.managed_umu_archive(UMU_LAUNCHER_VERSION)
+	if not _verified_archive(archive):
+		_download_archive(
+			archive,
+			timeout_seconds=timeout_seconds,
+			opener=opener,
+		)
+	_install_zipapp(archive, destination)
+	result = _result(destination, changed=True)
+	_write_json(
+		paths.managed_umu_metadata,
+		{
+			**asdict(result),
+			"installed_at": datetime.now().astimezone().isoformat(),
+		},
+	)
+	return result
+
+
+def _download_archive(
+	destination: Path,
+	*,
+	timeout_seconds: float,
+	opener: UmuOpener,
+) -> None:
+	request = Request(
+		UMU_LAUNCHER_URL,
+		headers={"User-Agent": USER_AGENT},
+		method="GET",
+	)
+	try:
+		response = opener(request, timeout=timeout_seconds)
+	except HTTPError as exc:
+		raise UmuBootstrapError(
+			f"portable UMU request failed: HTTP {exc.code}"
+		) from exc
+	except OSError as exc:
+		raise UmuBootstrapError(f"portable UMU request failed: {exc}") from exc
+
+	with response:
+		length = _content_length(response.headers.get("Content-Length"))
+		if length is not None and length > MAX_UMU_ARCHIVE_BYTES:
+			raise UmuBootstrapError(
+				f"portable UMU response is unexpectedly large: {length} bytes"
+			)
+		destination.parent.mkdir(parents=True, exist_ok=True)
+		temporary = _temporary_path(destination, "part")
+		digest = hashlib.sha256()
+		written = 0
+		try:
+			with temporary.open("xb") as handle:
+				while True:
+					chunk = response.read(READ_SIZE)
+					if not chunk:
+						break
+					written += len(chunk)
+					if written > MAX_UMU_ARCHIVE_BYTES:
+						raise UmuBootstrapError(
+							"portable UMU response exceeded the 4 MiB safety limit"
+						)
+					digest.update(chunk)
+					handle.write(chunk)
+				handle.flush()
+				os.fsync(handle.fileno())
+			if length is not None and written != length:
+				raise UmuBootstrapError(
+					f"incomplete portable UMU response: {written}/{length} bytes"
+				)
+			actual_hash = digest.hexdigest()
+			if actual_hash != UMU_ARCHIVE_SHA256:
+				raise UmuBootstrapError(
+					"portable UMU archive failed SHA-256 verification: "
+					f"expected {UMU_ARCHIVE_SHA256}, received {actual_hash}"
+				)
+			temporary.replace(destination)
+		except BaseException:
+			temporary.unlink(missing_ok=True)
+			raise
+
+
+def _install_zipapp(archive: Path, destination: Path) -> None:
+	destination.parent.mkdir(parents=True, exist_ok=True)
+	temporary = _temporary_path(destination, "tmp")
+	try:
+		try:
+			with tarfile.open(archive, "r:") as bundle:
+				member = bundle.getmember(UMU_ARCHIVE_MEMBER)
+				if not member.isfile():
+					raise UmuBootstrapError(
+						"portable UMU archive member is not a regular file"
+					)
+				if member.size <= 0 or member.size > MAX_UMU_EXECUTABLE_BYTES:
+					raise UmuBootstrapError(
+						f"portable UMU executable has invalid size: {member.size} bytes"
+					)
+				source = bundle.extractfile(member)
+				if source is None:
+					raise UmuBootstrapError(
+						"portable UMU executable could not be read from its archive"
+					)
+				with source, temporary.open("xb") as handle:
+					digest = hashlib.sha256()
+					written = 0
+					while True:
+						chunk = source.read(READ_SIZE)
+						if not chunk:
+							break
+						written += len(chunk)
+						digest.update(chunk)
+						handle.write(chunk)
+					handle.flush()
+					os.fsync(handle.fileno())
+		except (KeyError, OSError, tarfile.TarError) as exc:
+			raise UmuBootstrapError(f"invalid portable UMU archive: {exc}") from exc
+		if written != member.size or digest.hexdigest() != UMU_EXECUTABLE_SHA256:
+			raise UmuBootstrapError(
+				"portable UMU executable failed content verification"
+			)
+		if not zipfile.is_zipfile(temporary):
+			raise UmuBootstrapError("portable UMU executable is not a valid zipapp")
+		with zipfile.ZipFile(temporary) as application:
+			if "__main__.py" not in application.namelist():
+				raise UmuBootstrapError(
+					"portable UMU zipapp has no __main__.py entry point"
+				)
+			bad_member = application.testzip()
+			if bad_member is not None:
+				raise UmuBootstrapError(
+					f"portable UMU zipapp contains a corrupt member: {bad_member}"
+				)
+		temporary.chmod(0o755)
+		temporary.replace(destination)
+	except BaseException:
+		temporary.unlink(missing_ok=True)
+		raise
+
+
+def _verified_archive(path: Path) -> bool:
+	try:
+		return (
+			path.is_file()
+			and not path.is_symlink()
+			and path.stat().st_size <= MAX_UMU_ARCHIVE_BYTES
+			and _sha256(path) == UMU_ARCHIVE_SHA256
+		)
+	except OSError:
+		return False
+
+
+def _result(path: Path, *, changed: bool) -> UmuBootstrapResult:
+	return UmuBootstrapResult(
+		path=str(path),
+		version=UMU_LAUNCHER_VERSION,
+		source_url=UMU_LAUNCHER_URL,
+		archive_sha256=UMU_ARCHIVE_SHA256,
+		executable_sha256=UMU_EXECUTABLE_SHA256,
+		changed=changed,
+	)
+
+
+def _content_length(value: str | None) -> int | None:
+	if value is None:
+		return None
+	try:
+		length = int(value)
+	except ValueError as exc:
+		raise UmuBootstrapError(f"invalid portable UMU Content-Length: {value!r}") from exc
+	if length < 0:
+		raise UmuBootstrapError(f"invalid portable UMU Content-Length: {value!r}")
+	return length
+
+
+def _sha256(path: Path) -> str:
+	digest = hashlib.sha256()
+	with path.open("rb") as handle:
+		for chunk in iter(lambda: handle.read(READ_SIZE), b""):
+			digest.update(chunk)
+	return digest.hexdigest()
+
+
+def _temporary_path(path: Path, suffix: str) -> Path:
+	return path.with_name(
+		f".{path.name}.{os.getpid()}.{threading.get_ident()}."
+		f"{secrets.token_hex(8)}.{suffix}"
+	)
+
+
+def _write_json(path: Path, value: object) -> None:
+	path.parent.mkdir(parents=True, exist_ok=True)
+	temporary = _temporary_path(path, "tmp")
+	try:
+		with temporary.open("x", encoding="utf-8", newline="\n") as handle:
+			json.dump(value, handle, indent=2, sort_keys=True)
+			handle.write("\n")
+			handle.flush()
+			os.fsync(handle.fileno())
+		temporary.replace(path)
+	except BaseException:
+		temporary.unlink(missing_ok=True)
+		raise