pristontale-eu-linux/ptlinux/cli.py

"""Command-line interface for installation, updates, repair, and play."""

from __future__ import annotations

import argparse
from contextlib import AbstractContextManager, nullcontext
from dataclasses import asdict, replace
from datetime import datetime
import json
from pathlib import Path
import re
import signal
import sys
import threading
import time
from typing import Callable, Sequence
from uuid import uuid4

from ptdiag.logs import (
	LogFileState,
	classify_asset_references,
	newest_current_native_log,
	parse_current_run_log,
	snapshot_native_logs,
)

from . import __version__
from .command import CommandError, CommandOutcome, CommandRunner
from .container_host import (
	ContainerHostReport,
	container_host_error,
	inspect_container_host,
	repair_container_host,
	repair_container_host_after_update,
)
from .depot import (
	DownloadOptions,
	SyncProgress,
	read_cached_depot_launch_catalog,
	read_current_depot_file_catalog,
	preflight_managed_game,
	select_depot_start_option,
	sync_managed_game,
)
from .doctor import inspect_host
from .download import OFFICIAL_LAUNCHER_URL, fetch_official_launcher
from .importer import GameImportResult, discover_existing_game, import_existing_game
from .import_state import game_import_requires_verification, mark_game_import_verified
from .launcher import (
	LauncherError,
	WineTools,
	configure_launcher_prefix,
	find_wine_tools,
	launch_official_updater,
	prepare_launcher_binary,
)
from .locking import (
	LauncherPlayLease,
	launcher_operation_lock,
	launcher_play_lock,
	managed_game_process_is_active,
)
from .login_diagnostics import inspect_login_environment
from .login_trace import DEFAULT_TRACE_TIMEOUT_SECONDS, LOGIN_TRACE_DISABLED_REASON
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,
	select_runtime,
)
from .settings import Settings, ensure_settings, load_settings, reset_settings
from .self_update import (
	check_for_launcher_update,
	install_launcher_update,
	rollback_launcher_update,
	update_check_to_dict,
	verify_launcher_update_channel,
)
from .update_resolver import STABLE_CHANNEL
from .telemetry import (
	TelemetryService,
	classify_network_failure,
	current_telemetry,
	initialize_shared_telemetry,
	network_http_status,
	shutdown_shared_telemetry,
)
from .system_identity import is_running_inside_steam_runtime
from .umu import (
	UmuBootstrapResult,
	find_umu_launcher,
	provision_umu,
	provision_umu_for_startup,
)
from .window_diagnostics import (
	WINDOW_DIAGNOSTICS_ENV,
	WINDOW_DIAGNOSTICS_SESSION_ENV,
	parse_window_diagnostic,
)


def build_parser() -> argparse.ArgumentParser:
	parser = argparse.ArgumentParser(
		prog="pristontale-eu",
		description="PristonTale EU all-in-one Linux installer and clean launcher",
	)
	parser.add_argument(
		"--version",
		action="version",
		version=f"%(prog)s {__version__}",
	)
	parser.add_argument("--json", action="store_true", help="emit machine-readable output")
	subparsers = parser.add_subparsers(dest="command")

	subparsers.add_parser("gui", help="open the native Linux dashboard")
	subparsers.add_parser("doctor", help="check cross-distribution dependencies")
	container_host = subparsers.add_parser(
		"container-host",
		help="check Bubblewrap and Ubuntu AppArmor for the Steam Runtime",
	)
	container_host.add_argument(
		"--repair",
		action="store_true",
		help="repair only recognized Ubuntu AppArmor profile states",
	)
	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",
	)
	login_diagnostics.add_argument(
		"--timeout",
		type=float,
		default=5.0,
		help="TCP connection timeout in seconds (default: 5)",
	)
	login_diagnostics.add_argument(
		"--runtime",
		choices=("configured", "auto", "proton7", "proton7-container", "umu"),
		default="configured",
		help=(
			"probe one runtime without changing settings or starting the game "
			"(default: configured)"
		),
	)

	fetch = subparsers.add_parser("fetch", help="fetch and safely prepare the latest official launcher")
	fetch.add_argument("--dry-run", action="store_true")

	setup = subparsers.add_parser("setup", help="configure Wine and .NET Desktop for the updater")
	setup.add_argument("--dry-run", action="store_true")
	setup.add_argument("--force-runtime", action="store_true")

	update = subparsers.add_parser(
		"update",
		help="safely synchronize the official game depot without Wine",
	)
	update.add_argument("--dry-run", action="store_true")
	update.add_argument("--repair", action="store_true", help="hash-check every game file")
	import_options = update.add_mutually_exclusive_group()
	import_options.add_argument(
		"--seed-from",
		type=Path,
		help="import an existing game before updating",
	)
	import_options.add_argument(
		"--no-import",
		action="store_true",
		help="disable automatic existing-game discovery",
	)

	sync = subparsers.add_parser(
		"sync",
		help="native official depot synchronization without bootstrap discovery",
	)
	sync.add_argument("--dry-run", action="store_true")
	sync.add_argument("--repair", action="store_true", help="hash-check every game file")

	legacy_update = subparsers.add_parser(
		"wine-update",
		help="legacy official WPF updater (normally unnecessary)",
	)
	legacy_update.add_argument("--dry-run", action="store_true")
	legacy_update.add_argument("--force-runtime", action="store_true")
	legacy_import_options = legacy_update.add_mutually_exclusive_group()
	legacy_import_options.add_argument("--seed-from", type=Path)
	legacy_import_options.add_argument("--no-import", action="store_true")

	import_game = subparsers.add_parser(
		"import-game",
		help="safely seed the managed installation from an existing game",
	)
	import_game.add_argument("source", type=Path)
	import_game.add_argument("--dry-run", action="store_true")

	compatibility = subparsers.add_parser(
		"cache-compatibility",
		help="cache one exact verified Game.exe/game.dll compatibility pair",
	)
	compatibility.add_argument("source", type=Path)
	prepare = subparsers.add_parser("prepare", help="build the read-only runtime view and scene fix")
	prepare.add_argument("--game-dir", type=Path)

	play = subparsers.add_parser("play", help="launch the game in the managed Proton window")
	play.add_argument("--game-dir", type=Path)
	play.add_argument(
		"--realm",
		help="launch an available realm from the last official depot manifest",
	)
	play.add_argument("--no-box", action="store_true", help="launch without X11 window styling")
	play.add_argument(
		"--publisher-dll",
		"--no-world-fix",
		action="store_true",
		dest="publisher_dll",
		help=(
			"compatibility option retained for older scripts; publisher code is "
			"always used without the retired world fix"
		),
	)
	play.add_argument("--dry-run", action="store_true")
	play.add_argument("--skip-runtime-setup", action="store_true")
	login_trace = subparsers.add_parser(
		"login-trace",
		help="retired unsafe authentication trace (always refuses to launch)",
	)
	login_trace.add_argument("--game-dir", type=Path)
	login_trace.add_argument("--no-box", action="store_true", help="launch without X11 window styling")
	login_trace.add_argument("--skip-runtime-setup", action="store_true")
	login_trace.add_argument(
		"--timeout",
		type=float,
		default=DEFAULT_TRACE_TIMEOUT_SECONDS,
		help="maximum game runtime in seconds (default: 600)",
	)

	settings = subparsers.add_parser("settings", help="show, initialize, or reset settings")
	settings_action = settings.add_mutually_exclusive_group()
	settings_action.add_argument(
		"--init",
		action="store_true",
		help="create missing settings and migrate legacy files without resetting choices",
	)
	settings_action.add_argument(
		"--reset",
		action="store_true",
		help="explicitly replace all launcher settings with defaults",
	)

	launcher_update = subparsers.add_parser(
		"launcher-update",
		help="check, verify, install, or roll back the native Linux launcher",
	)
	launcher_action = launcher_update.add_mutually_exclusive_group()
	launcher_action.add_argument(
		"--check",
		action="store_true",
		help="authenticate the update manifest without installing it",
	)
	launcher_action.add_argument(
		"--verify",
		action="store_true",
		help="download and fully validate the published release without installing it",
	)
	launcher_action.add_argument(
		"--rollback",
		nargs="?",
		const="",
		metavar="VERSION",
		help=(
			"switch to VERSION from signed release history; without VERSION, "
			"switch to the locally retained previous release"
		),
	)
	launcher_update.add_argument(
		"--manifest-url",
		help="override the signed stable-manifest URL",
	)
	launcher_update.add_argument(
		"--public-key",
		type=Path,
		help="override the pinned launcher-update public key",
	)
	launcher_update.add_argument(
		"--history-url",
		help="override the signed launcher release-history URL",
	)
	return parser


def main(
	argv: Sequence[str] | None = None,
	*,
	_manage_telemetry: bool | None = None,
) -> int:
	parser = build_parser()
	arguments = parser.parse_args(argv)
	command = arguments.command or "gui"
	paths = AppPaths.from_environment()
	manage_telemetry = argv is None if _manage_telemetry is None else _manage_telemetry
	telemetry: TelemetryService | None = None
	telemetry_owner = False
	if manage_telemetry:
		interface = "gui" if command == "gui" else "cli"
		initialized_telemetry, telemetry_owner = initialize_shared_telemetry(
			paths,
			session_metadata={"interface": interface, "command": command},
			report_launcher_lifecycle=command == "gui",
		)
		telemetry = initialized_telemetry
	telemetry_events = _command_telemetry_events(command, arguments, paths)
	operation_started = time.monotonic()
	if telemetry is not None and telemetry_events is not None:
		operation, started_event, _success_event, _failed_event = telemetry_events
		if operation == "repair":
			telemetry.send_event(
				"repair_requested",
				metadata={"command": command, "reason": "user_requested"},
				context={"operation": operation, "stage": "requested"},
			)
		telemetry.send_event(
			started_event,
			metadata={"command": command},
			context={"operation": operation, "stage": "started"},
		)
	try:
		returncode = _dispatch_command(parser, arguments, command, paths)
	except KeyboardInterrupt:
		_report_command_failure(
			telemetry,
			telemetry_events,
			command,
			operation_started,
			"Operation interrupted",
			stage="interrupted",
		)
		message = "Interrupted; PristonTale and its compatibility runtime were stopped."
		if arguments.json:
			_print_json({"ok": False, "error": message, "interrupted": True})
		else:
			print(f"\n{message}", file=sys.stderr)
		return 130
	except (ValueError, OSError, RuntimeError) as exc:
		_report_command_failure(
			telemetry,
			telemetry_events,
			command,
			operation_started,
			str(exc),
			stage=_failure_stage(command, exc),
		)
		if telemetry is not None and isinstance(exc, ValueError):
			telemetry.send_event(
				"configuration_error",
				severity="error",
				message=str(exc),
				metadata={"setting": "unknown", "reason": "invalid_value"},
				context={"operation": command, "stage": "configuration"},
			)
		if arguments.json:
			_print_json({"ok": False, "error": str(exc)})
		else:
			print(f"Error: {exc}", file=sys.stderr)
			if isinstance(exc, CommandError):
				if exc.outcome.stdout:
					print(exc.outcome.stdout.rstrip(), file=sys.stderr)
				if exc.outcome.stderr:
					print(exc.outcome.stderr.rstrip(), file=sys.stderr)
		return 1
	except BaseException as exc:
		if telemetry is not None:
			telemetry.report_exception(
				"launcher_exception",
				exc,
				metadata={"command": command},
				context={"operation": command, "stage": "dispatch"},
				severity="critical",
			)
		_report_command_failure(
			telemetry,
			telemetry_events,
			command,
			operation_started,
			str(exc),
			stage="dispatch",
		)
		raise
	else:
		_report_command_result(
			telemetry,
			telemetry_events,
			command,
			operation_started,
			returncode,
		)
		return returncode
	finally:
		if telemetry is not None and telemetry_owner:
			shutdown_shared_telemetry(telemetry)


def _dispatch_command(
	parser: argparse.ArgumentParser,
	arguments: argparse.Namespace,
	command: str,
	paths: AppPaths,
) -> int:
	if command == "gui":
		provision_umu_for_startup(paths)
		_container_host_startup_migration(paths)
		from .ui import run_dashboard

		run_dashboard(paths)
		return 0
	if command == "doctor":
		return _doctor(arguments.json)
	if command == "container-host":
		return _container_host(arguments.repair, arguments.json)
	if command == "runtime-bootstrap":
		return _runtime_bootstrap(paths, arguments.json)
	if command == "login-diagnostics":
		return _login_diagnostics(
			paths,
			arguments.timeout,
			arguments.runtime,
			arguments.json,
		)
	if command == "fetch":
		return _fetch(paths, arguments.dry_run, arguments.json)
	if command == "setup":
		return _setup(
			paths,
			arguments.dry_run,
			arguments.force_runtime,
			arguments.json,
		)
	if command == "update":
		return _native_update(
			paths,
			arguments.dry_run,
			arguments.repair,
			arguments.seed_from,
			arguments.no_import,
			arguments.json,
		)
	if command == "sync":
		return _native_update(
			paths,
			arguments.dry_run,
			arguments.repair,
			None,
			True,
			arguments.json,
		)
	if command == "wine-update":
		return _wine_update(
			paths,
			arguments.dry_run,
			arguments.force_runtime,
			arguments.seed_from,
			arguments.no_import,
			arguments.json,
		)
	if command == "import-game":
		return _import_game(
			paths,
			arguments.source,
			arguments.dry_run,
			arguments.json,
		)
	if command == "cache-compatibility":
		result = cache_compatibility_binaries(paths, arguments.source)
		_emit({"ok": True, "compatibility": asdict(result)}, arguments.json)
		return 0
	if command == "prepare":
		return _prepare(paths, arguments.game_dir, arguments.json)
	if command == "play":
		return _play(
			paths,
			arguments.game_dir,
			arguments.no_box,
			arguments.dry_run,
			arguments.skip_runtime_setup,
			arguments.json,
			realm=arguments.realm,
		)
	if command == "login-trace":
		return _login_trace(
			paths,
			arguments.game_dir,
			arguments.no_box,
			arguments.skip_runtime_setup,
			arguments.timeout,
			arguments.json,
		)
	if command == "settings":
		return _settings(paths, arguments.init, arguments.reset, arguments.json)
	if command == "launcher-update":
		return _launcher_update(
			paths,
			arguments.check,
			arguments.verify,
			arguments.rollback,
			arguments.manifest_url,
			arguments.public_key,
			arguments.json,
			history_url=arguments.history_url,
		)
	parser.error(f"unknown command: {command}")


def _command_telemetry_events(
	command: str,
	arguments: argparse.Namespace,
	paths: AppPaths,
) -> tuple[str, str, str, str] | None:
	if bool(getattr(arguments, "dry_run", False)):
		return None
	if command in {"update", "sync", "wine-update"}:
		if bool(getattr(arguments, "repair", False)):
			return ("repair", "repair_started", "repair_success", "repair_failed")
		if not paths.source_game.exists():
			return ("install", "install_started", "install_success", "install_failed")
		return ("update", "update_started", "update_success", "update_failed")
	if command == "launcher-update" and not (
		bool(getattr(arguments, "check", False))
		or bool(getattr(arguments, "verify", False))
		or getattr(arguments, "rollback", None) is not None
	):
		return (
			"launcher_update",
			"update_started",
			"update_success",
			"update_failed",
		)
	if command in {"setup", "runtime-bootstrap"}:
		return (
			"runtime_install",
			"runtime_install_started",
			"runtime_install_success",
			"runtime_install_failed",
		)
	if command == "doctor":
		return (
			"compatibility_check",
			"compatibility_check_started",
			"compatibility_check_completed",
			"compatibility_check_failed",
		)
	return None


def _report_command_result(
	telemetry: TelemetryService | None,
	events: tuple[str, str, str, str] | None,
	command: str,
	started_at: float,
	returncode: int,
) -> None:
	if telemetry is None or events is None:
		return
	operation, _started_event, success_event, failed_event = events
	duration_ms = round((time.monotonic() - started_at) * 1000)
	telemetry.send_event(
		success_event if returncode == 0 else failed_event,
		severity="info" if returncode == 0 else "error",
		message=None if returncode == 0 else f"Command exited with status {returncode}",
		metadata={
			"command": command,
			"duration_ms": duration_ms,
			"exit_code": returncode,
		},
		context={
			"operation": operation,
			"stage": "completed" if returncode == 0 else "command_exit",
		},
	)


def _report_command_failure(
	telemetry: TelemetryService | None,
	events: tuple[str, str, str, str] | None,
	command: str,
	started_at: float,
	message: str,
	*,
	stage: str = "failed",
) -> None:
	if telemetry is None:
		return
	if command == "play":
		telemetry.send_event(
			"game_launch_failed",
			severity="error",
			message=message,
			metadata={
				"duration_ms": round((time.monotonic() - started_at) * 1000),
			},
			context={"operation": "launch_game", "stage": "preflight"},
		)
		return
	if events is None:
		return
	operation, _started_event, _success_event, failed_event = events
	telemetry.send_event(
		failed_event,
		severity="error",
		message=message,
		metadata={
			"command": command,
			"duration_ms": round((time.monotonic() - started_at) * 1000),
		},
		context={"operation": operation, "stage": stage},
	)


def _failure_stage(command: str, exception: BaseException) -> str:
	if isinstance(exception, ValueError):
		return "configuration"
	if isinstance(exception, CommandError):
		return "subprocess"
	message = str(exception).casefold()
	if any(value in message for value in ("manifest", "cdn", "request failed")):
		return "manifest"
	if any(value in message for value in ("download", "incomplete response")):
		return "download"
	if any(value in message for value in ("sha-1", "checksum", "hash mismatch")):
		return "validation"
	if any(value in message for value in ("permission", "read-only", "cannot write")):
		return "filesystem"
	if any(value in message for value in ("prefix", "wine", "proton", "umu")):
		return "runtime_check"
	return f"{command}_operation"


def _doctor(as_json: bool) -> int:
	report = inspect_host()
	if as_json:
		_print_json(report.to_dict())
	else:
		print(f"Distribution: {report.distribution} {report.version} ({report.architecture})")
		print(f"Display:      {report.display_backend}")
		for tool in report.tools:
			status = tool.path or "MISSING"
			required = "required" if tool.required else "optional"
			print(f"{tool.name:12} {status} [{required}] — {tool.purpose}")
		optional_missing = [
			tool.name for tool in report.tools if not tool.required and tool.path is None
		]
		if optional_missing:
			print(
				"Optional tools missing (these do not block launch): "
				+ ", ".join(optional_missing)
			)
		print(f"Host readiness: {'READY' if report.ok else 'BLOCKED'}")
		if report.blockers:
			print("Blocking diagnostics:")
			for blocker in report.blockers:
				print(f"  - {blocker}")
		for warning in report.warnings:
			print(f"Warning: {warning}")
		if report.package_hint and any(
			tool.required and tool.path is None for tool in report.tools
		):
			print(report.package_hint)
	return 0 if report.ok else 1


def _container_host(repair: bool, as_json: bool) -> int:
	report = repair_container_host() if repair else inspect_container_host()
	ok = _container_host_ok(report)
	payload = {"ok": ok, "container_host": report.to_dict()}
	if as_json:
		_print_json(payload)
	else:
		status = "ready" if report.ready else "blocked"
		print(f"Bubblewrap: {status} ({report.bwrap_path or 'not found'})")
		if report.active_profile_source:
			print(f"AppArmor:   {report.active_profile_source}")
		elif report.apparmor_restrict_unprivileged_userns == "1":
			print(
				"AppArmor:   unprivileged user namespaces are restricted; "
				"no bwrap profile found"
			)
		if report.repair_actions:
			for action in report.repair_actions:
				print(f"Repair:     {action}")
		if report.repair_error:
			print(f"Repair failed: {report.repair_error}", file=sys.stderr)
		elif report.recommendation:
			print(f"Action: {report.recommendation}", file=sys.stderr)
	return 0 if ok else 1


def _container_host_ok(report: ContainerHostReport) -> bool:
	return (
		report.ready
		and not report.broken_umu_profile_link
		and report.repair_error is None
	)


def _container_host_startup_migration(paths: AppPaths) -> None:
	report = repair_container_host_after_update(paths)
	if report.repair_attempted and not _container_host_ok(report):
		print(f"Warning: {container_host_error(report)}", file=sys.stderr)


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,
	runtime_choice: str,
	as_json: bool,
) -> int:
	settings = _load_or_default_settings(paths, persist=False)
	runner = _runner(paths, "login-diagnostics", False)
	with _operation_guard(paths, "login-diagnostics", False):
		report = inspect_login_environment(
			paths,
			settings,
			timeout_seconds=timeout_seconds,
			runner=runner,
			runtime_override=(
				None if runtime_choice == "configured" else runtime_choice
			),
		)
	if as_json:
		_print_json(report.to_dict())
	else:
		probe = report.login_server
		print(f"Launcher:      {report.launcher_version} ({report.launcher_module})")
		print(f"Python:        {report.python_executable}")
		print(f"Host:          {report.distribution_id} ({report.kernel_release})")
		print(
			"Runtime test:  "
			f"{report.runtime_requested} "
			f"(configured: {report.runtime_configured}; settings unchanged)"
		)
		status = "REACHABLE" if probe.reachable else "UNREACHABLE"
		latency = f", {probe.latency_ms:.1f} ms" if probe.latency_ms is not None else ""
		print(f"Login server:  {status} ({probe.host}:{probe.port}{latency})")
		if probe.error:
			print(f"Network error: {probe.error}")
		print("Authentication: not tested; no credentials or game-protocol data were sent")
		print(f"Login responses: {report.login_response_ownership}")
		identity_status = (
			"ready"
			if report.client_identity_ready is True
			else "NOT READY"
			if report.client_identity_ready is False
			else "not probed"
		)
		print(f"Client identity: {identity_status}; raw identity values were not collected")
		print(f"Privacy:       {report.client_identity_privacy}")
		if not report.login_trace_enabled:
			print("Login trace:    retired (will not launch or instrument the game)")
		print(f"Publisher test: {report.publisher_comparison_command}")
		print(f"Game profile:  {report.game_profile}")
		print(f"Game.exe:      {report.game_exe}")
		if report.game_exe_sha256:
			print(f"Game SHA-256:  {report.game_exe_sha256}")
		print(f"game.dll:      {report.game_dll}")
		if report.game_dll_sha256:
			print(f"DLL SHA-256:   {report.game_dll_sha256}")
		print(
			"Managed source: "
			+ ("matches runtime" if report.runtime_matches_managed_source else "DIFFERS")
		)
		if report.depot_build is not None:
			print(
				f"Depot:         build {report.depot_build} "
				f"({report.depot_version or 'unknown version'})"
			)
		if report.runtime_kind:
			print(f"Runtime:       {report.runtime_kind} ({report.runtime_path or 'unknown path'})")
		else:
			print(f"Runtime:       unavailable ({report.runtime_error or 'selection failed'})")
		if report.proton_sync_effective is not None:
			print(
				"Proton sync:   "
				f"{report.proton_sync_configured} -> {report.proton_sync_effective}"
			)
		prefix_status = "ready" if report.prefix_ready else "not ready"
		directx_status = "ready" if report.directx_ready else "not ready"
		print(f"Prefix:        {prefix_status}; DirectX helpers {directx_status}")
		print(f"Prefix path:   {report.prefix or 'unavailable'}")
		print(f"Latest log:    {report.latest_game_log or 'not found'}")
		if report.latest_game_log_modified_at:
			print(f"Log modified:  {report.latest_game_log_modified_at}")
		for warning in report.warnings:
			print(f"Warning: {warning}")
	return 0 if report.preflight_ok else 1


def _fetch(paths: AppPaths, dry_run: bool, as_json: bool) -> int:
	if not dry_run:
		paths.create_base_directories()
	if dry_run:
		value = {
			"ok": True,
			"dry_run": True,
			"url": OFFICIAL_LAUNCHER_URL,
			"download": str(paths.launcher_pristine),
			"prepared": str(paths.launcher_patched),
		}
	else:
		telemetry = current_telemetry()
		download_started_at = time.monotonic()
		if telemetry is not None:
			telemetry.send_event(
				"download_started",
				metadata={
					"resource_type": "launcher",
					"mirror_id": "pristontale.eu",
				},
				context={"operation": "download", "stage": "launcher_download"},
			)
		try:
			download = fetch_official_launcher(
				paths.launcher_pristine,
				paths.launcher_metadata,
			)
		except BaseException as exc:
			if telemetry is not None:
				telemetry.send_event(
					"download_failed",
					severity="error",
					message=str(exc),
					metadata={
						"resource_type": "launcher",
						"duration_ms": round(
							(time.monotonic() - download_started_at) * 1000
						),
						"failure_type": classify_network_failure(exc),
						"http_status": network_http_status(exc),
						"mirror_id": "pristontale.eu",
					},
					context={"operation": "download", "stage": "launcher_download"},
				)
			raise
		download_duration = max(0.0, time.monotonic() - download_started_at)
		download_bytes = download.size if download.changed else 0
		if telemetry is not None:
			telemetry.send_event(
				"download_completed",
				metadata={
					"resource_type": "launcher",
					"download_bytes": download_bytes,
					"content_bytes": download.size,
					"duration_ms": round(download_duration * 1000),
					"average_mbps": (
						round((download_bytes * 8) / download_duration / 1_000_000, 3)
						if download_bytes and download_duration > 0
						else None
					),
					"retry_count": 0,
					"resume_used": False,
					"http_status": 200 if download.changed else 304,
					"changed": download.changed,
					"mirror_id": "pristontale.eu",
				},
				context={"operation": "download", "stage": "launcher_download"},
			)
		validation_started_at = time.monotonic()
		try:
			patch = prepare_launcher_binary(
				paths.launcher_pristine,
				paths.launcher_patched,
			)
		except BaseException as exc:
			if telemetry is not None:
				telemetry.send_event(
					"file_validation_failed",
					severity="error",
					message=str(exc),
					metadata={
						"file_category": "launcher",
						"duration_ms": round(
							(time.monotonic() - validation_started_at) * 1000
						),
					},
					context={"operation": "download", "stage": "validation"},
				)
			raise
		value = {
			"ok": True,
			"download": asdict(download),
			"patch": asdict(patch),
			"warning": None if patch.supported_profile else (
				"The launcher is newer than the verified font-patch profile; "
				"it was retained byte-for-byte."
			),
		}
	_emit(value, as_json)
	return 0


def _setup(
	paths: AppPaths,
	dry_run: bool,
	force_runtime: bool,
	as_json: bool,
) -> int:
	tools = find_wine_tools()
	if tools is None:
		raise LauncherError("Wine, Wineboot, and Winetricks are required; run doctor")
	runner = _runner(paths, "setup", dry_run)
	with _operation_guard(paths, "setup", dry_run):
		results = _configure_launcher_runtime(
			paths,
			tools,
			runner,
			force_runtime=force_runtime,
		)
	_emit(
		{
			"ok": True,
			"dry_run": dry_run,
			"prefix": str(paths.launcher_prefix),
			"commands": [asdict(result) for result in results],
		},
		as_json,
	)
	return 0


def _native_update(
	paths: AppPaths,
	dry_run: bool,
	repair: bool,
	seed_from: Path | None,
	no_import: bool,
	as_json: bool,
) -> int:
	"""Use the native verified depot client for normal installs and updates."""

	settings = _load_or_default_settings(paths, persist=not dry_run)
	download_options = _configured_download_options(settings)
	runner = _runner(paths, "native-update", dry_run)
	with _operation_guard(paths, "native-update", dry_run):
		_print_operation_status(as_json, 1, 3, "Checking local game files…")
		game_import = None
		if not no_import:
			game_import = _maybe_import_game(
				paths,
				settings,
				seed_from,
				runner,
				announce=not as_json,
			)
		verify_import = game_import is not None or game_import_requires_verification(paths)
		_print_operation_status(as_json, 2, 3, "Synchronizing official game files…")
		result = sync_managed_game(
			paths,
			repair=repair or verify_import,
			quarantine_unmanaged=verify_import,
			dry_run=dry_run,
			progress=None if as_json else _print_sync_progress,
			**(
				{"download_options": download_options}
				if download_options is not None
				else {}
			),
		)
		if verify_import and not dry_run:
			mark_game_import_verified(
				paths,
				build=result.build,
				files_quarantined=result.files_quarantined,
				quarantine_directory=result.quarantine_directory,
			)
		overlay = None
		if not dry_run and paths.source_game.is_dir():
			_print_operation_status(as_json, 3, 3, "Preparing the clean game runtime view…")
			overlay = asdict(prepare_runtime_view(paths, paths.source_game))
	_emit(
		{
			"ok": True,
			"dry_run": dry_run,
			"game_import": asdict(game_import) if game_import else None,
			"native_update": asdict(result),
			"runtime_view": overlay,
		},
		as_json,
	)
	return 0


def _wine_update(
	paths: AppPaths,
	dry_run: bool,
	force_runtime: bool,
	seed_from: Path | None,
	no_import: bool,
	as_json: bool,
) -> int:
	_fetch(paths, dry_run, True)
	tools = find_wine_tools()
	if tools is None:
		raise LauncherError("Wine, Wineboot, and Winetricks are required; run doctor")
	runner = _runner(paths, "update", dry_run)
	settings = _load_or_default_settings(paths, persist=not dry_run)
	download_options = _configured_download_options(settings)
	with _operation_guard(paths, "update", dry_run):
		_configure_launcher_runtime(
			paths,
			tools,
			runner,
			force_runtime=force_runtime,
		)
		game_import = None
		if not no_import:
			game_import = _maybe_import_game(
				paths,
				settings,
				seed_from,
				runner,
				announce=not as_json,
			)
		updater = launch_official_updater(paths, tools, runner)
		native_verification = None
		verify_import = not dry_run and (
			game_import is not None or game_import_requires_verification(paths)
		)
		if verify_import:
			native_verification = sync_managed_game(
				paths,
				repair=True,
				quarantine_unmanaged=True,
				progress=None if as_json else _print_sync_progress,
				**(
					{"download_options": download_options}
					if download_options is not None
					else {}
				),
			)
			mark_game_import_verified(
				paths,
				build=native_verification.build,
				files_quarantined=native_verification.files_quarantined,
				quarantine_directory=native_verification.quarantine_directory,
			)
		overlay = None
		if not dry_run and paths.source_game.is_dir():
			overlay = asdict(prepare_runtime_view(paths, paths.source_game))
	value = {
		"ok": True,
		"dry_run": dry_run,
		"game_import": asdict(game_import) if game_import else None,
		"updater": asdict(updater),
		"native_verification": (
			asdict(native_verification) if native_verification is not None else None
		),
		"runtime_view": overlay,
	}
	_emit(value, as_json)
	return 0


def _configure_launcher_runtime(
	paths: AppPaths,
	tools: WineTools,
	runner: CommandRunner,
	*,
	force_runtime: bool,
) -> list[CommandOutcome]:
	if not force_runtime or runner.dry_run:
		return configure_launcher_prefix(
			paths,
			tools,
			runner,
			force_runtime=force_runtime,
		)
	telemetry = current_telemetry()
	started_at = time.monotonic()
	metadata = {
		"runtime_type": "dotnet_desktop",
		"runtime_version": "6",
	}
	if telemetry is not None:
		telemetry.send_event(
			"runtime_reinstall_requested",
			metadata={**metadata, "reason": "force_runtime"},
			context={"operation": "runtime_repair", "stage": "requested"},
		)
		telemetry.send_event(
			"runtime_repair_started",
			metadata=metadata,
			context={"operation": "runtime_repair", "stage": "started"},
		)
	try:
		results = configure_launcher_prefix(
			paths,
			tools,
			runner,
			force_runtime=True,
		)
	except BaseException as exc:
		if telemetry is not None:
			telemetry.send_event(
				"runtime_repair_failed",
				severity="error",
				message=str(exc),
				metadata={
					**metadata,
					"duration_ms": round((time.monotonic() - started_at) * 1000),
				},
				context={"operation": "runtime_repair", "stage": "runtime_install"},
			)
		raise
	if telemetry is not None:
		telemetry.send_event(
			"runtime_repair_success",
			metadata={
				**metadata,
				"duration_ms": round((time.monotonic() - started_at) * 1000),
			},
			context={"operation": "runtime_repair", "stage": "completed"},
		)
	return results


def _print_sync_progress(progress: SyncProgress) -> None:
	"""Emit line-delimited progress that the native Tk dashboard can consume."""

	print(f"PROGRESS {json.dumps(asdict(progress), sort_keys=True)}", flush=True)


def _configured_download_options(settings: Settings) -> DownloadOptions | None:
	if (
		settings.download_speed_limit_bytes_per_second is None
		and not settings.low_bandwidth_mode
	):
		return None
	return DownloadOptions(
		bandwidth_limit_bytes_per_second=(
			settings.download_speed_limit_bytes_per_second
		),
		low_bandwidth=settings.low_bandwidth_mode,
	)


def _print_operation_status(
	as_json: bool,
	step: int,
	total: int,
	message: str,
) -> None:
	"""Emit a structured operation stage for the native launcher dashboard."""

	if not as_json:
		print(
			f"STATUS {json.dumps({'step': step, 'total': total, 'message': message})}",
			flush=True,
		)


def _import_game(
	paths: AppPaths,
	source: Path,
	dry_run: bool,
	as_json: bool,
) -> int:
	runner = _runner(paths, "import-game", dry_run)
	if not as_json:
		print(f"Importing existing game from {source.expanduser()}…", flush=True)
	with _operation_guard(paths, "import-game", dry_run):
		result = _import_with_telemetry(paths, source, runner)
	_emit({"ok": True, "dry_run": dry_run, "game_import": asdict(result)}, as_json)
	return 0


def _maybe_import_game(
	paths: AppPaths,
	settings: Settings,
	seed_from: Path | None,
	runner: CommandRunner,
	*,
	announce: bool,
) -> GameImportResult | None:
	if paths.source_game.exists() or paths.source_game.is_symlink():
		if seed_from is not None:
			raise RuntimeError(
				f"cannot seed an existing managed game directory: {paths.source_game}"
			)
		return None
	candidate = seed_from
	if candidate is None:
		candidate = discover_existing_game(paths, settings)
	if candidate is None:
		return None
	if announce:
		print(f"Bootstrapping from existing game: {candidate}", flush=True)
	return _import_with_telemetry(paths, candidate, runner)


def _import_with_telemetry(
	paths: AppPaths,
	source: Path,
	runner: CommandRunner,
) -> GameImportResult:
	telemetry = current_telemetry()
	started_at = time.monotonic()
	if telemetry is not None and not runner.dry_run:
		telemetry.send_event(
			"import_started",
			context={"operation": "import", "stage": "copy"},
		)
	try:
		result = import_existing_game(paths, source, runner)
	except BaseException as exc:
		if telemetry is not None and not runner.dry_run:
			telemetry.send_event(
				"import_failed",
				severity="error",
				message=str(exc),
				metadata={
					"duration_ms": round((time.monotonic() - started_at) * 1000),
				},
				context={"operation": "import", "stage": "copy"},
			)
		raise
	if telemetry is not None and not runner.dry_run:
		telemetry.send_event(
			"import_success",
			metadata={
				"files_imported": result.files,
				"bytes_imported": result.bytes,
				"duration_ms": round((time.monotonic() - started_at) * 1000),
			},
			context={"operation": "import", "stage": "completed"},
		)
	return result


def _prepare(paths: AppPaths, game_dir: Path | None, as_json: bool) -> int:
	with _operation_guard(paths, "prepare", False):
		if game_dir is None and game_import_requires_verification(paths):
			raise LauncherError(
				"the imported game has not completed its full publisher verification; "
				"run pristontale-eu update"
			)
		settings = ensure_settings(paths)
		source = _source_game(paths, settings, game_dir)
		result = prepare_runtime_view(paths, source)
	_emit({"ok": True, "runtime_view": asdict(result)}, as_json)
	return 0


def _play(
	paths: AppPaths,
	game_dir: Path | None,
	no_box: bool,
	dry_run: bool,
	skip_runtime_setup: bool,
	as_json: bool,
	*,
	realm: str | None = None,
) -> int:
	with _play_guard(paths, dry_run) as play_lease:
		return _play_locked(
			paths,
			game_dir,
			no_box,
			dry_run,
			skip_runtime_setup,
			as_json,
			realm=realm,
			reuse_prepared=not play_lease.preparation_allowed,
			before_game=play_lease.allow_shared_clients,
			game_lease_fds=play_lease.inherited_fds,
		)


def _play_locked(
	paths: AppPaths,
	game_dir: Path | None,
	no_box: bool,
	dry_run: bool,
	skip_runtime_setup: bool,
	as_json: bool,
	*,
	realm: str | None = None,
	reuse_prepared: bool = False,
	before_game: Callable[[], None] | None = None,
	game_lease_fds: Sequence[int] = (),
) -> int:
	telemetry = current_telemetry()
	launch_requested_at = time.monotonic()
	window_diagnostics_active = (
		telemetry is not None
		and not dry_run
		and bool(getattr(telemetry, "enabled", True))
	)
	launch_session_id = uuid4().hex if window_diagnostics_active else None
	launch_session_metadata = (
		{"launch_session_id": launch_session_id}
		if launch_session_id is not None
		else {}
	)
	if telemetry is not None and not dry_run:
		telemetry.send_event(
			"game_launch_requested",
			metadata=launch_session_metadata,
			context={"operation": "launch_game", "stage": "preflight"},
		)
	if game_dir is None and game_import_requires_verification(paths):
		raise LauncherError(
			"the imported game has not completed its full publisher verification; "
			"run pristontale-eu update before playing"
		)
	settings = _load_or_default_settings(paths, persist=not dry_run)
	launcher_nested_in_steam_runtime = is_running_inside_steam_runtime()
	if launcher_nested_in_steam_runtime and not dry_run:
		print(
			"Warning: the launcher itself is running inside Steam Runtime. Start "
			"it from the desktop menu or a host terminal; nesting its UMU runtime "
			"can cause black game output.",
			file=sys.stderr,
			flush=True,
		)
	window_launch_metadata: dict[str, object] = {
		"gamebox_enabled": not no_box,
		"resizable_window_requested": settings.allow_window_resizing,
		"launcher_inside_steam_runtime": launcher_nested_in_steam_runtime,
	}
	if (
		launcher_nested_in_steam_runtime
		and settings.allow_window_resizing
		and not no_box
	):
		window_launch_metadata["gamescope_disabled_reason"] = (
			"launcher_inside_steam_runtime"
		)
	start_option = select_depot_start_option(
		read_cached_depot_launch_catalog(paths.depot_launch_catalog_cache),
		realm or settings.realm,
	)
	source = _source_game(paths, settings, game_dir)
	overlay = None
	depot_file_catalog = None
	depot_preflight = None
	if game_dir is None and not dry_run:
		depot_file_catalog = read_current_depot_file_catalog(paths)
	_print_operation_status(
		as_json,
		1,
		4,
		(
			"Reusing the active managed game view…"
			if reuse_prepared
			else "Preparing the managed game view…"
		),
	)
	if reuse_prepared:
		if game_dir is not None:
			raise LauncherError(
				"an additional client cannot replace the active managed game view; "
				"close every client before using --game-dir"
			)
		if not paths.runtime_view.is_dir():
			raise LauncherError(
				"another game client is active, but the prepared runtime view is missing"
			)
	elif not dry_run:
		if game_dir is None:
			if depot_file_catalog is not None:
				depot_preflight = preflight_managed_game(
					paths,
					catalog=depot_file_catalog,
				)
				if depot_preflight is not None and not depot_preflight.ready:
					details = "; ".join(
						f"{issue.path}: {issue.reason}"
						for issue in depot_preflight.issues[:3]
					)
					raise LauncherError(
						"the managed game integrity check found "
						f"{depot_preflight.issue_count} missing or changed file(s)"
						+ (f" ({details})" if details else "")
						+ "; run `pristontale-eu update --repair` before playing"
					)
		overlay = prepare_runtime_view(paths, source)
	_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:
		_print_operation_status(
			as_json,
			3,
			4,
			"Reusing the active Proton and DirectX runtime…",
		)
		if not game_runtime_ready(paths, selection):
			raise LauncherError(
				"another game client is active, but its prepared runtime no longer "
				"matches the current settings; close every client and launch again"
			)
	elif not skip_runtime_setup:
		_print_operation_status(as_json, 3, 4, "Preparing Proton and DirectX components…")
		setup_results = prepare_game_runtime(paths, selection, runner)
	else:
		_print_operation_status(as_json, 3, 4, "Using the existing Proton runtime…")
	_print_operation_status(as_json, 4, 4, "Starting the PristonTale game client…")
	native_log_directory = paths.game_save / "logs"
	native_log_baseline = snapshot_native_logs(
		native_log_directory,
		patterns=("log_*.txt",),
	)
	crash_directory = paths.game_save / "crash"
	crash_baseline = snapshot_native_logs(
		crash_directory,
		patterns=("*.dat", "*.dmp"),
	)
	game_started_ns = time.time_ns()
	game_started_at = time.monotonic()
	launch_success_lock = threading.Lock()
	launch_success_reported = False
	reported_runtime_warnings: set[str] = set()
	if telemetry is not None and not dry_run:
		telemetry.send_event(
			"game_launch_started",
			metadata={
				**launch_session_metadata,
				**window_launch_metadata,
				"launch_duration_ms": round(
					(game_started_at - launch_requested_at) * 1000
				),
				"launch_stage": "game_start",
				"runtime": selection.kind,
				"realm": start_option.name,
			},
			context={"operation": "launch_game", "stage": "game_start"},
		)
	if before_game is not None:
		before_game()
	managed_asset_paths = (
		[entry.path for entry in depot_file_catalog.files]
		if depot_file_catalog is not None
		else None
	)

	def observe_game_output(line: str) -> str | None:
		"""Timestamp launch success when the supervisor first confirms readiness."""

		nonlocal launch_success_reported
		presentation_failure = _gamescope_presentation_failure(line)
		if telemetry is not None and not dry_run and presentation_failure is not None:
			with launch_success_lock:
				if presentation_failure not in reported_runtime_warnings:
					reported_runtime_warnings.add(presentation_failure)
					telemetry.send_event(
						"game_window_diagnostic",
						severity="warning",
						metadata={
							**launch_session_metadata,
							**window_launch_metadata,
							"component": "gamescope_host",
							"diagnostic": presentation_failure,
							"runtime": selection.kind,
							"realm": start_option.name,
						},
						context={
							"operation": "launch_game",
							"stage": "presentation",
						},
					)
		window_diagnostic = parse_window_diagnostic(line)
		if window_diagnostic is not None:
			diagnostic_session_id = window_diagnostic.get("session_id")
			if launch_session_id is None or diagnostic_session_id != launch_session_id:
				return line
			if telemetry is not None and window_diagnostics_active:
				with launch_success_lock:
					telemetry.send_event(
						"game_window_diagnostic",
						severity=_window_diagnostic_severity(window_diagnostic),
						metadata={
							"runtime": selection.kind,
							"realm": start_option.name,
							**window_diagnostic,
						},
						context={
							"operation": "launch_game",
							"stage": "window_diagnostics",
						},
					)
			return None
		live_readiness_signal = _game_readiness_signal(line)
		if telemetry is None or dry_run or live_readiness_signal is None:
			return line
		with launch_success_lock:
			if launch_success_reported:
				return line
			ready_at = time.monotonic()
			launch_success_reported = telemetry.send_event(
				"game_launch_success",
				metadata={
					**launch_session_metadata,
					**window_launch_metadata,
					"launch_duration_ms": round(
						(ready_at - launch_requested_at) * 1000
					),
					"game_start_duration_ms": round(
						(ready_at - game_started_at) * 1000
					),
					"game_window_ready": True,
					"readiness_signal": live_readiness_signal,
					"runtime": selection.kind,
					"realm": start_option.name,
				},
				context={"operation": "launch_game", "stage": "ready"},
			)
		return line

	try:
		outcome = run_game(
			paths,
			settings,
			selection,
			runner,
			start_option=start_option,
			use_gamebox=not no_box,
			concurrent_prefix_launch=reuse_prepared,
			check=False,
			output_line_filter=observe_game_output,
			extra_environment=(
				{
					WINDOW_DIAGNOSTICS_ENV: "1",
					WINDOW_DIAGNOSTICS_SESSION_ENV: launch_session_id,
				}
				if launch_session_id is not None
				else None
			),
			pass_fds=game_lease_fds,
		)
	except KeyboardInterrupt:
		if telemetry is not None and not dry_run:
			telemetry.send_event(
				"game_exited",
				severity="warning",
				message="Game launch was interrupted",
				metadata={
					**launch_session_metadata,
					**window_launch_metadata,
					"runtime_seconds": round(time.monotonic() - game_started_at),
					"clean_exit": False,
					"session_duration_class": _session_duration_class(
						time.monotonic() - game_started_at
					),
				},
				context={"operation": "launch_game", "stage": "running"},
			)
		current_game_log = _current_game_log(
			native_log_directory,
			native_log_baseline,
			game_started_ns,
			managed_asset_paths=managed_asset_paths,
			source_game=source if managed_asset_paths is not None else None,
		)
		if current_game_log is not None and not as_json:
			_print_interrupted_game_log(current_game_log)
		raise
	current_game_log = _current_game_log(
		native_log_directory,
		native_log_baseline,
		game_started_ns,
		managed_asset_paths=managed_asset_paths,
		source_game=source if managed_asset_paths is not None else None,
	)
	current_game_crash = _current_game_crash(
		crash_directory,
		crash_baseline,
		game_started_ns,
	)
	runtime_issues = _runtime_issue_summary(outcome.stderr)
	game_window_ready = runtime_issues["game_window_ready"] is True
	if (
		selection.kind == "proton7-container"
		and runtime_issues["host_futex_fatal_count"]
	):
		runtime_issues["recommended_one_variable_comparison"] = "proton7"
	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"
	clean_exit = outcome.returncode == 0 and current_game_crash is None
	launch_succeeded = (
		clean_exit or game_window_ready or current_game_log is not None
	)
	report = {
		"ok": clean_exit,
		"launch_succeeded": launch_succeeded,
		"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,
		"depot_preflight": (
			{
				**asdict(depot_preflight),
				"ready": depot_preflight.ready,
				"fully_verified": depot_preflight.fully_verified,
			}
			if depot_preflight
			else None
		),
		"setup_commands": [asdict(result) for result in setup_results],
		"game": asdict(outcome),
		"current_game_log": current_game_log,
		"current_game_crash": current_game_crash,
		"runtime_issues": runtime_issues,
	}
	game_runtime_seconds = max(0.0, time.monotonic() - game_started_at)
	if telemetry is not None and not dry_run:
		common_game_metadata: dict[str, object] = {
			**launch_session_metadata,
			**window_launch_metadata,
			"runtime_seconds": round(game_runtime_seconds),
			"clean_exit": clean_exit,
			"timed_out": outcome.timed_out,
			"launch_succeeded": launch_succeeded,
			"game_window_ready": game_window_ready,
			"crash_dump": current_game_crash is not None,
			"session_duration_class": _session_duration_class(game_runtime_seconds),
			"runtime": selection.kind,
			"realm": start_option.name,
			**_game_exit_metadata(outcome.returncode, selection.kind),
		}
		if launch_succeeded:
			if not launch_success_reported:
				readiness_signal = _game_readiness_signal(outcome.stderr)
				if readiness_signal is None:
					readiness_signal = (
						"game_log"
						if current_game_log is not None
						else "clean_process_exit"
					)
				telemetry.send_event(
					"game_launch_success",
					metadata={
						**launch_session_metadata,
						**window_launch_metadata,
						"detection_duration_ms": round(game_runtime_seconds * 1000),
						"detected_after_exit": True,
						"game_window_ready": game_window_ready,
						"readiness_signal": readiness_signal,
						"runtime": selection.kind,
						"realm": start_option.name,
					},
					context={"operation": "launch_game", "stage": "post_exit"},
				)
		else:
			telemetry.send_event(
				"game_launch_failed",
				severity="error",
				message=(
					"Game client wrote a crash dump"
					if current_game_crash is not None
					else "Game session timed out"
					if outcome.timed_out
					else f"Game client exited with status {outcome.returncode}"
				),
				metadata={
					**common_game_metadata,
					"runtime_issues": runtime_issues,
					"stdout": outcome.stdout,
					"stderr": outcome.stderr,
				},
				context={"operation": "launch_game", "stage": "game_start"},
			)
		if launch_succeeded:
			exit_message = None
			if current_game_crash is not None:
				exit_message = "Game client wrote a crash dump"
			elif outcome.timed_out:
				exit_message = "Game session timed out"
			elif common_game_metadata.get("signal_inferred") is True:
				exit_message = (
					"Game session ended with status "
					f"{common_game_metadata['raw_exit_code']}, consistent with "
					f"{common_game_metadata['signal_name']} wrapped by UMU"
				)
			elif "signal" in common_game_metadata:
				exit_message = (
					"Game session was terminated by "
					f"{common_game_metadata['signal_name']}"
				)
			elif outcome.returncode not in {0, None}:
				exit_message = f"Game session ended with status {outcome.returncode}"
			telemetry.send_event(
				"game_exited",
				severity=(
					"error"
					if current_game_crash is not None
					else "info" if clean_exit else "warning"
				),
				message=exit_message,
				metadata=(
					common_game_metadata
					if clean_exit
					else {
						**common_game_metadata,
						"runtime_issues": runtime_issues,
						"stdout": outcome.stdout,
						"stderr": outcome.stderr,
					}
				),
				context={"operation": "launch_game", "stage": "exited"},
			)
	if as_json or dry_run:
		_emit(report, as_json)
	else:
		_print_game_outcome(
			outcome,
			current_game_log,
			runtime_issues,
			current_game_crash,
		)
	return 0 if outcome.returncode == 0 and current_game_crash is None else 1


def _session_duration_class(duration_seconds: float) -> str:
	if duration_seconds < 30:
		return "under_30_seconds"
	if duration_seconds < 5 * 60:
		return "under_5_minutes"
	return "normal"


def _game_exit_metadata(
	returncode: int | None,
	runtime_kind: str,
) -> dict[str, object]:
	metadata: dict[str, object] = {"exit_code": returncode}
	if returncode is None:
		return metadata
	if returncode < 0:
		signal_number = -returncode
		metadata.update(
			{
				"exit_code": 128 + min(signal_number, 127),
				"raw_exit_code": returncode,
				"signal": signal_number,
				"signal_name": _signal_name(signal_number),
			}
		)
		return metadata

	# UMU returns Popen's negative child status from a Python main function.
	# SystemExit encodes -SIGKILL as 247 before our Python gamebox sees it.
	wrapped_sigkill = 256 - int(signal.SIGKILL)
	if runtime_kind in {"umu", "proton7-container"} and returncode == wrapped_sigkill:
		metadata.update(
			{
				"exit_code": 128 + int(signal.SIGKILL),
				"raw_exit_code": returncode,
				"signal": int(signal.SIGKILL),
				"signal_name": signal.SIGKILL.name,
				"signal_inferred": True,
				"signal_inference": "umu_python_system_exit",
			}
		)
	return metadata


def _signal_name(signal_number: int) -> str:
	try:
		return signal.Signals(signal_number).name
	except ValueError:
		return f"signal {signal_number}"


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],
	started_ns: int,
	*,
	managed_asset_paths: Sequence[str] | None = None,
	source_game: Path | None = None,
) -> dict[str, object] | None:
	state = newest_current_native_log(
		directory,
		baseline,
		started_ns,
		patterns=("log_*.txt",),
	)
	if state is None:
		return None
	analysis = parse_current_run_log(state, baseline)
	asset_references = classify_asset_references(
		analysis,
		managed_asset_paths,
		source_game=source_game,
	)
	return {
		"path": str(state.path),
		"modified_at": datetime.fromtimestamp(
			state.mtime_ns / 1_000_000_000
		).astimezone().isoformat(),
		"size": state.size,
		"load_texture_to_material_failed": analysis.load_texture_to_material_failed,
		"import_scene_threw": analysis.import_scene_threw,
		"asset_pack_health_failed": analysis.asset_pack_health_failed,
		"managed_asset_integrity_failed": asset_references.managed_integrity_failed,
		"asset_references": asset_references.to_dict(),
		"runtime_out_of_memory": analysis.runtime_out_of_memory,
		"out_of_memory_count": len(analysis.out_of_memory_events),
		"out_of_memory_effect_paths": analysis.out_of_memory_effect_paths,
		"effect_creation_failures": [
			event.to_dict() for event in analysis.effect_creation_failures
		],
		"failed_import_scenes": analysis.failed_import_scenes,
		"stage_without_new_stage": analysis.stage_without_new_stage,
		"module_code_sync_error": analysis.module_code_sync_error,
		"module_code_sync_left": analysis.module_code_sync_left,
		"module_code_sync_right": analysis.module_code_sync_right,
		"module_code_sync_delta": analysis.module_code_sync_delta,
		"banned_code_1": analysis.banned_code_1,
		"banned_code_1_in_native_log": analysis.banned_code_1,
		"banned_code_1_observation_scope": (
			"native-game-log-only; server-provided login-screen text is not "
			"written to this log"
		),
		"failed_file_find_count": analysis.failed_file_find_count,
		"failed_file_load_count": analysis.failed_file_load_count,
		"failed_file_read_count": analysis.failed_file_read_count,
		"failed_asset_file_count": analysis.failed_asset_file_count,
		"failed_texture_file_count": analysis.failed_texture_file_count,
		"failed_effect_file_count": analysis.failed_effect_file_count,
		"missing_texture_reference_count": analysis.missing_texture_reference_count,
		"missing_effect_reference_count": analysis.missing_effect_reference_count,
		"missing_image_reference_count": analysis.missing_image_reference_count,
		"missing_pack_image_reference_count": analysis.missing_pack_image_reference_count,
		"stage_without_new_stage_events": [
			event.to_dict()
			for event in analysis.events
			if event.kind == "stage_without_new_stage"
		],
		"scene_imports": {
			name: [scene.to_dict() for scene in imports]
			for name, imports in sorted(analysis.scenes.items())
		},
		"failed_file_count": len(analysis.failed_files),
		"failed_files": [event.to_dict() for event in analysis.failed_files[:64]],
	}


def _current_game_crash(
	directory: Path,
	baseline: dict[str, LogFileState],
	started_ns: int,
) -> dict[str, object] | None:
	state = newest_current_native_log(
		directory,
		baseline,
		started_ns,
		patterns=("*.dat", "*.dmp"),
	)
	if state is None:
		return None
	return {
		"path": str(state.path),
		"modified_at": datetime.fromtimestamp(
			state.mtime_ns / 1_000_000_000
		).astimezone().isoformat(),
		"size": state.size,
	}


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"
	bwrap_uid_map_denied = "bwrap: setting up uid map: permission denied"
	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 {
		"game_window_ready": _game_window_was_ready(folded),
		"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,
		"gamescope_present_queue_unsupported": (
			"physical device queue doesn't support presenting on our surface" in folded
		),
		"gamescope_vulkan_output_failed_count": folded.count(
			"vulkan_make_output failed"
		),
		"steam_api_app_id_missing": "steamapi_init() failed; no appid found" in folded,
		"dbghelp_symdeferred_assertion_count": folded.count(dbghelp_assertion),
		"bubblewrap_uid_map_permission_denied": bwrap_uid_map_denied in folded,
	}


def _gamescope_presentation_failure(output: str) -> str | None:
	folded = output.casefold()
	for diagnostic, marker in (
		(
			"gamescope_present_queue_unsupported",
			"physical device queue doesn't support presenting on our surface",
		),
		("gamescope_vulkan_output_failed", "vulkan_make_output failed"),
	):
		if marker in folded:
			return diagnostic
	return None


def _game_window_was_ready(stderr: str) -> bool:
	return _game_readiness_signal(stderr) is not None


def _game_readiness_signal(output: str) -> str | None:
	folded = output.casefold()
	for signal_name, marker in (
		("window_ready", "game window ready:"),
		("rendered_frame", "rendered game frame:"),
		(
			"window_supervision",
			"game window resizing is disabled to prevent repeated direct3d device resets",
		),
	):
		if marker in folded:
			return signal_name
	return None


def _window_diagnostic_severity(diagnostic: dict[str, object]) -> str:
	name = diagnostic.get("diagnostic")
	details = diagnostic.get("details")
	if isinstance(name, str) and (
		name.endswith(("_failed", "_unavailable"))
		or name in {
			"host_hidden_state_cleared",
			"host_inactive_remap",
			"host_minimize_latched",
		}
	):
		return "warning"
	if isinstance(details, dict):
		if name == "host_scaling_configured" and not all(
			details.get(key) is True
			for key in (
				"resize_increments_applied",
				"scaling_enabled",
				"compositing_preferred",
			)
		):
			return "warning"
		if name == "scaling_configured" and details.get("enabled") is not True:
			return "warning"
	return "debug"


def _print_game_outcome(
	outcome: CommandOutcome,
	current_game_log: dict[str, object] | None,
	runtime_issues: dict[str, object],
	current_game_crash: dict[str, object] | None = None,
) -> 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 current_game_crash is not None:
		crash_path = current_game_crash.get("path")
		crash_path_text = crash_path if isinstance(crash_path, str) else "unknown"
		print(
			f"Game client crashed and wrote a dump: {crash_path_text}",
			file=sys.stderr,
		)
	elif outcome.timed_out:
		print("Game client timed out and was stopped.", file=sys.stderr)
	elif outcome.returncode == 0:
		print("Game client exited normally.")
	elif outcome.returncode is None:
		print("Game client exited without a status.", file=sys.stderr)
	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,
		)
	if runtime_issues["bubblewrap_uid_map_permission_denied"]:
		print(
			"Ubuntu denied Bubblewrap's UID map before Proton started. Run "
			"`pristontale-eu container-host --repair` and retry.",
			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:
	log_path = report.get("path")
	log_path_text = log_path if isinstance(log_path, str) else "unknown"
	print(f"\nCurrent game log: {log_path_text}", file=sys.stderr)
	asset_references = report.get("asset_references")
	catalog_available = (
		asset_references.get("catalog_available")
		if isinstance(asset_references, dict)
		else False
	)
	if report["module_code_sync_error"]:
		print(
			"Detected a publisher module-integrity mismatch; the game was launched "
			"with publisher-pristine code in this launcher version.",
			file=sys.stderr,
		)
	elif report["runtime_out_of_memory"]:
		effects = report.get("out_of_memory_effect_paths")
		effect_text = ""
		if isinstance(effects, list) and effects:
			effect_text = " while creating " + ", ".join(str(path) for path in effects[:3])
		print(
			"Detected Direct3D resource exhaustion in the 32-bit game process"
			+ effect_text
			+ ". Restarting the client can recover temporarily; depot repair will not "
			"fix this failure.",
			file=sys.stderr,
		)
	elif report["managed_asset_integrity_failed"]:
		print(
			"The game reported an asset that belongs to the publisher manifest but is "
			"missing from the managed installation.",
			file=sys.stderr,
		)
		print(
			"Run `pristontale-eu update --repair` and restart.",
			file=sys.stderr,
		)
	elif report["asset_pack_health_failed"] and not catalog_available:
		print(
			"The game reported many unresolved asset paths, but no matching cached "
			"publisher manifest was available to classify them. Run "
			"`pristontale-eu update` to refresh the manifest cache.",
			file=sys.stderr,
		)
	elif report["load_texture_to_material_failed"]:
		print(
			"Detected a texture/material failure while loading the game world.",
			file=sys.stderr,
		)
	elif report["import_scene_threw"] or report["failed_import_scenes"]:
		print("Detected a game-world scene import failure.", file=sys.stderr)


def _login_trace(
	paths: AppPaths,
	game_dir: Path | None,
	no_box: bool,
	skip_runtime_setup: bool,
	timeout_seconds: float,
	as_json: bool,
) -> int:
	# Preserve the command name so old instructions fail safely before touching
	# the prefix, preparing the runtime view, or starting Game.exe.
	_ = paths, game_dir, no_box, skip_runtime_setup, timeout_seconds, as_json
	raise RuntimeError(LOGIN_TRACE_DISABLED_REASON)


def _settings(
	paths: AppPaths,
	initialize: bool,
	reset: bool,
	as_json: bool,
) -> int:
	if reset:
		settings = reset_settings(paths)
	elif initialize:
		settings = ensure_settings(paths)
	else:
		settings = load_settings(paths)
	_emit(
		{
			"ok": True,
			"path": str(paths.settings_file),
			"settings": asdict(settings),
		},
		as_json,
	)
	return 0


def _launcher_update(
	paths: AppPaths,
	check_only: bool,
	verify: bool,
	rollback: str | bool | None,
	manifest_url: str | None,
	public_key: Path | None,
	as_json: bool,
	*,
	history_url: str | None = None,
) -> int:
	try:
		selected_channel = load_settings(paths).update_channel
	except (OSError, TypeError, ValueError):
		# Keep update recovery usable when settings are absent or damaged.
		selected_channel = STABLE_CHANNEL
	rollback_requested = rollback is not None and rollback is not False
	if rollback_requested:
		target_version = rollback if isinstance(rollback, str) and rollback else None
		result = rollback_launcher_update(
			paths,
			target_version=target_version,
			selected_channel=selected_channel,
			history_url=history_url,
			public_key_path=public_key,
		)
		_emit(
			{
				"ok": True,
				"launcher_rollback": asdict(result),
			},
			as_json,
		)
		return 0
	if check_only:
		check = check_for_launcher_update(
			manifest_url=manifest_url,
			public_key_path=public_key,
			channel=selected_channel,
		)
		_emit(
			{
				"ok": True,
				"launcher_update": update_check_to_dict(check),
			},
			as_json,
		)
		return 0
	if verify:
		_print_operation_status(as_json, 1, 2, "Authenticating the launcher release…")
		result = verify_launcher_update_channel(
			manifest_url=manifest_url,
			public_key_path=public_key,
			channel=selected_channel,
		)
		_print_operation_status(as_json, 2, 2, "Published launcher release verified.")
		_emit(
			{
				"ok": True,
				"launcher_update_verification": asdict(result),
			},
			as_json,
		)
		return 0
	_print_operation_status(as_json, 1, 4, "Verifying the signed launcher release…")
	result = install_launcher_update(
		paths,
		manifest_url=manifest_url,
		public_key_path=public_key,
		channel=selected_channel,
	)
	_print_operation_status(
		as_json,
		2,
		4,
		"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,
		4,
		"Checking Steam Runtime sandbox permissions (authentication may be requested)…",
	)
	container_report = repair_container_host()
	container_warning = container_report.repair_error
	if not container_report.ready and container_warning is None:
		container_warning = container_report.recommendation
	if container_warning and not as_json:
		print(f"Warning: {container_warning}", file=sys.stderr)
	_print_operation_status(as_json, 4, 4, "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,
			},
			"container_host": {
				**container_report.to_dict(),
				"warning": container_warning,
			},
		},
		as_json,
	)
	return 0


def _source_game(
	paths: AppPaths,
	_settings: Settings,
	argument: Path | None,
) -> Path:
	"""Select the managed game unless an explicit diagnostic path was supplied.

	The persisted ``game_source`` setting is an import seed only. It must not
	redirect normal Play, Prepare, or Login Trace operations away from the
	launcher-managed tree.
	"""

	if argument is not None:
		return argument.expanduser().resolve()
	return paths.source_game


def _load_or_default_settings(paths: AppPaths, *, persist: bool) -> Settings:
	if paths.settings_file.is_file():
		return load_settings(paths)
	if persist:
		return ensure_settings(paths)
	return Settings()


def _runner(paths: AppPaths, operation: str, dry_run: bool) -> CommandRunner:
	timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S-%f")
	return CommandRunner(
		paths.logs / f"{timestamp}-{operation}.jsonl",
		dry_run=dry_run,
		operation=operation,
	)


def _play_guard(
	paths: AppPaths,
	dry_run: bool,
) -> AbstractContextManager[LauncherPlayLease]:
	if dry_run:
		return nullcontext(LauncherPlayLease.dry_run())
	return launcher_play_lock(
		paths.launcher_operation_lock,
		active_game_probe=lambda: _managed_game_process_is_active(paths),
	)


def _operation_guard(
	paths: AppPaths,
	operation: str,
	dry_run: bool,
) -> AbstractContextManager[None]:
	if dry_run:
		return nullcontext()
	return launcher_operation_lock(
		paths.launcher_operation_lock,
		operation,
		active_game_probe=lambda: _managed_game_process_is_active(paths),
	)


def _managed_game_process_is_active(paths: AppPaths) -> bool:
	return managed_game_process_is_active(
		paths.runtime_view,
		(paths.proton_prefix, paths.umu_prefix),
	)


def _emit(value: object, as_json: bool) -> None:
	if as_json:
		_print_json(value)
	else:
		if isinstance(value, dict) and value.get("ok") is True:
			print("Done.")
		print(json.dumps(value, indent=2, sort_keys=True))


def _print_json(value: object) -> None:
	print(json.dumps(value, indent=2, sort_keys=True))