From f06fc2e4e3a6c531c675c05dc68d37ac21abcbca Mon Sep 17 00:00:00 2001 From: m4ko Date: Mon, 7 Sep 2026 21:49:37 +0200 Subject: [PATCH] Added telemetry system to get information about issues. --- ptlinux/__main__.py | 29 +- ptlinux/cli.py | 713 ++++++++++-- ptlinux/command.py | 24 +- ptlinux/container_host.py | 73 +- ptlinux/depot.py | 470 +++++++- ptlinux/importer.py | 1 + ptlinux/launcher.py | 11 +- ptlinux/paths.py | 6 + ptlinux/runtime.py | 6 + ptlinux/self_update.py | 65 +- ptlinux/settings.py | 2 + ptlinux/telemetry.py | 1541 +++++++++++++++++++++++++ ptlinux/telemetry_environment.py | 1221 ++++++++++++++++++++ ptlinux/ui/dashboard.py | 8 +- ptlinux/ui/dialogs/settings_dialog.py | 49 +- ptlinux/umu.py | 75 +- 16 files changed, 4125 insertions(+), 169 deletions(-) create mode 100644 ptlinux/telemetry.py create mode 100644 ptlinux/telemetry_environment.py diff --git a/ptlinux/__main__.py b/ptlinux/__main__.py index 04ff322..ccac227 100644 --- a/ptlinux/__main__.py +++ b/ptlinux/__main__.py @@ -13,10 +13,17 @@ def main() -> int: repair_container_host_after_update, ) from .paths import AppPaths + from .telemetry import initialize_shared_telemetry, shutdown_shared_telemetry from .umu import provision_umu_for_startup + telemetry = None + telemetry_owner = False try: paths = AppPaths.from_environment() + telemetry, telemetry_owner = initialize_shared_telemetry( + paths, + session_metadata={"interface": "gui", "command": "gui"}, + ) # 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) @@ -36,13 +43,33 @@ def main() -> int: except KeyboardInterrupt: return 130 except (ValueError, OSError, RuntimeError) as exc: + if telemetry is not None: + telemetry.report_exception( + "launcher_exception", + exc, + metadata={"command": "gui"}, + context={"operation": "launcher", "stage": "startup"}, + ) print(f"Error: {exc}", file=sys.stderr) return 1 + except BaseException as exc: + if telemetry is not None: + telemetry.report_exception( + "launcher_exception", + exc, + metadata={"command": "gui"}, + context={"operation": "launcher", "stage": "startup"}, + severity="critical", + ) + raise + finally: + if telemetry is not None and telemetry_owner: + shutdown_shared_telemetry(telemetry) return 0 from .cli import main as cli_main - return cli_main(arguments) + return cli_main(arguments, _manage_telemetry=True) raise SystemExit(main()) diff --git a/ptlinux/cli.py b/ptlinux/cli.py index b910c78..e353ab5 100644 --- a/ptlinux/cli.py +++ b/ptlinux/cli.py @@ -45,6 +45,7 @@ from .importer import GameImportResult, discover_existing_game, import_existing_ 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, @@ -75,6 +76,14 @@ from .self_update import ( update_check_to_dict, verify_launcher_update_channel, ) +from .telemetry import ( + TelemetryService, + classify_network_failure, + current_telemetry, + initialize_shared_telemetry, + network_http_status, + shutdown_shared_telemetry, +) from .umu import ( UmuBootstrapResult, find_umu_launcher, @@ -266,114 +275,55 @@ def build_parser() -> argparse.ArgumentParser: return parser -def main(argv: Sequence[str] | None = None) -> int: +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}, + ) + telemetry = initialized_telemetry + if command != "gui": + initialized_telemetry.mark_ready( + {"interface": interface, "command": command} + ) + 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: - 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, - ) - parser.error(f"unknown command: {command}") + 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}) @@ -381,6 +331,22 @@ def main(argv: Sequence[str] | None = None) -> int: 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: @@ -391,6 +357,269 @@ def main(argv: Sequence[str] | None = None) -> int: 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, + ) + 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 bool(getattr(arguments, "rollback", False)) + ): + 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: @@ -568,14 +797,84 @@ def _fetch(paths: AppPaths, dry_run: bool, as_json: bool) -> int: "prepared": str(paths.launcher_patched), } else: - download = fetch_official_launcher( - paths.launcher_pristine, - paths.launcher_metadata, - ) - patch = prepare_launcher_binary( - paths.launcher_pristine, - paths.launcher_patched, - ) + 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), @@ -600,7 +899,7 @@ def _setup( 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_prefix( + results = _configure_launcher_runtime( paths, tools, runner, @@ -696,7 +995,7 @@ def _wine_update( 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_prefix( + _configure_launcher_runtime( paths, tools, runner, @@ -751,6 +1050,69 @@ def _wine_update( 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.""" @@ -796,7 +1158,7 @@ def _import_game( if not as_json: print(f"Importing existing game from {source.expanduser()}…", flush=True) with _operation_guard(paths, "import-game", dry_run): - result = import_existing_game(paths, source, runner) + result = _import_with_telemetry(paths, source, runner) _emit({"ok": True, "dry_run": dry_run, "game_import": asdict(result)}, as_json) return 0 @@ -822,7 +1184,46 @@ def _maybe_import_game( return None if announce: print(f"Bootstrapping from existing game: {candidate}", flush=True) - return import_existing_game(paths, candidate, runner) + 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: @@ -877,6 +1278,13 @@ def _play_locked( before_game: Callable[[], None] | None = None, game_lease_fds: Sequence[int] = (), ) -> int: + telemetry = current_telemetry() + launch_requested_at = time.monotonic() + if telemetry is not None and not dry_run: + telemetry.send_event( + "game_launch_requested", + 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; " @@ -970,6 +1378,20 @@ def _play_locked( patterns=("*.dat", "*.dmp"), ) game_started_ns = time.time_ns() + game_started_at = time.monotonic() + if telemetry is not None and not dry_run: + telemetry.send_event( + "game_launch_started", + 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 = ( @@ -990,6 +1412,20 @@ def _play_locked( 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={ + "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, @@ -1044,6 +1480,41 @@ def _play_locked( "current_game_crash": current_game_crash, "runtime_issues": runtime_issues, } + game_runtime_seconds = max(0.0, time.monotonic() - game_started_at) + clean_exit = outcome.returncode == 0 and current_game_crash is None + if telemetry is not None and not dry_run: + common_game_metadata = { + "runtime_seconds": round(game_runtime_seconds), + "exit_code": outcome.returncode, + "clean_exit": clean_exit, + "session_duration_class": _session_duration_class(game_runtime_seconds), + "runtime": selection.kind, + "realm": start_option.name, + } + if clean_exit: + telemetry.send_event( + "game_launch_success", + metadata=common_game_metadata, + context={"operation": "launch_game", "stage": "completed"}, + ) + else: + telemetry.send_event( + "game_launch_failed", + severity="error", + message=( + "Game client wrote a crash dump" + if current_game_crash is not None + else f"Game client exited with status {outcome.returncode}" + ), + metadata={**common_game_metadata, "runtime_issues": runtime_issues}, + context={"operation": "launch_game", "stage": "running"}, + ) + telemetry.send_event( + "game_exited", + severity="info" if clean_exit else "warning", + metadata=common_game_metadata, + context={"operation": "launch_game", "stage": "exited"}, + ) if as_json or dry_run: _emit(report, as_json) else: @@ -1056,6 +1527,14 @@ def _play_locked( 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 _ensure_portable_umu_for_play( paths: AppPaths, settings: Settings, @@ -1504,7 +1983,11 @@ def _load_or_default_settings(paths: AppPaths, *, persist: bool) -> 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) + return CommandRunner( + paths.logs / f"{timestamp}-{operation}.jsonl", + dry_run=dry_run, + operation=operation, + ) def _play_guard( diff --git a/ptlinux/command.py b/ptlinux/command.py index 5b42351..20c9d6b 100644 --- a/ptlinux/command.py +++ b/ptlinux/command.py @@ -15,6 +15,8 @@ import threading import time from typing import Callable, Mapping, Sequence, TextIO +from .telemetry import current_telemetry + OutputLineFilter = Callable[[str], str | None] MAX_CAPTURE_CHARACTERS = 1024 * 1024 @@ -85,9 +87,16 @@ class CommandError(RuntimeError): class CommandRunner: """Run argument arrays with timeouts and append-only JSONL logging.""" - def __init__(self, log_path: Path, *, dry_run: bool = False) -> None: + def __init__( + self, + log_path: Path, + *, + dry_run: bool = False, + operation: str | None = None, + ) -> None: self.log_path = log_path self.dry_run = dry_run + self.operation = operation def run( self, @@ -100,6 +109,7 @@ class CommandRunner: stream_output: bool = False, output_line_filter: OutputLineFilter | None = None, pass_fds: Sequence[int] = (), + telemetry_stage: str | None = None, ) -> CommandOutcome: arguments = [str(value) for value in argv] if not arguments: @@ -169,6 +179,18 @@ class CommandRunner: dry_run=False, ) self._record(outcome, cwd, overrides) + if outcome.timed_out or outcome.returncode != 0: + telemetry = current_telemetry() + if telemetry is not None: + telemetry.report_subprocess_failure( + "subprocess_failed", + outcome, + command=outcome.argv[:1], + metadata={ + "operation": self.operation or "subprocess", + **({"stage": telemetry_stage} if telemetry_stage else {}), + }, + ) if check and (outcome.timed_out or outcome.returncode != 0): raise CommandError(outcome) return outcome diff --git a/ptlinux/container_host.py b/ptlinux/container_host.py index c453fbc..0436e98 100644 --- a/ptlinux/container_host.py +++ b/ptlinux/container_host.py @@ -12,6 +12,7 @@ import re import subprocess import sys import tempfile +import time from typing import Callable, Mapping, Sequence from . import __version__ @@ -301,16 +302,30 @@ def repair_container_host( files=files, command_executor=command_executor, ) + started_at = time.monotonic() + _send_dependency_fix_event( + "dependency_fix_requested", + metadata={"reason": "container_host", "issue": before.issue}, + stage="requested", + ) + _send_dependency_fix_event( + "dependency_fix_started", + metadata={"issue": before.issue, "repairable": before.repairable}, + stage="started", + ) if before.ready and not before.broken_umu_profile_link: + _send_dependency_fix_outcome(before, started_at) return before if not before.repairable: - return replace( + result = replace( before, repair_attempted=True, repair_error=( before.recommendation or "No safe automatic repair is available." ), ) + _send_dependency_fix_outcome(result, started_at) + return result actions: list[str] = [] created_managed_profile = False profile_loaded = False @@ -372,12 +387,14 @@ def repair_container_host( error = str(exc) if rollback_error: error = f"{error}; rollback also failed: {rollback_error}" - return replace( + result = replace( after_failure, repair_attempted=True, repair_actions=tuple(actions), repair_error=error, ) + _send_dependency_fix_outcome(result, started_at) + return result after = inspect_container_host( environment=environment, @@ -391,13 +408,63 @@ def repair_container_host( "still fails. " + (after.recommendation or "Inspect the host security policy.") ) - return replace( + result = replace( after, repair_attempted=True, repaired=after.ready, repair_actions=tuple(actions), repair_error=error, ) + _send_dependency_fix_outcome(result, started_at) + return result + + +def _send_dependency_fix_outcome( + report: ContainerHostReport, + started_at: float, +) -> None: + success = report.ready and not report.broken_umu_profile_link and ( + report.repair_error is None + ) + _send_dependency_fix_event( + "dependency_fix_success" if success else "dependency_fix_failed", + severity="info" if success else "error", + message=None if success else report.repair_error, + metadata={ + "duration_ms": round((time.monotonic() - started_at) * 1000), + "ready": report.ready, + "repaired": report.repaired, + "repair_attempted": report.repair_attempted, + "action_count": len(report.repair_actions), + "issue": report.issue, + }, + stage="completed" if success else "repair", + ) + + +def _send_dependency_fix_event( + event: str, + *, + severity: str = "info", + message: str | None = None, + metadata: Mapping[str, object] | None = None, + stage: str, +) -> None: + try: + from .telemetry import current_telemetry + + telemetry = current_telemetry() + if telemetry is not None: + telemetry.send_event( + event, + severity=severity, + message=message, + metadata=metadata, + context={"operation": "dependency_fix", "stage": stage}, + ) + except Exception: + # Host recovery must remain independent from optional telemetry. + pass def repair_container_host_after_update(paths: AppPaths) -> ContainerHostReport: diff --git a/ptlinux/depot.py b/ptlinux/depot.py index 5ae6320..d60cebe 100644 --- a/ptlinux/depot.py +++ b/ptlinux/depot.py @@ -25,6 +25,7 @@ from uuid import uuid4 from . import __version__ from .paths import AppPaths +from .telemetry import classify_network_failure, current_telemetry CDN_CONFIGURATION_URL = "https://cdn-main.chatplex.io/pristontale-eu/cdn_configuration.json" @@ -93,9 +94,16 @@ class DepotError(RuntimeError): class _RetryableDownloadError(DepotError): """One transient network or content-integrity failure.""" - def __init__(self, message: str, *, discard_partial: bool = False) -> None: + def __init__( + self, + message: str, + *, + discard_partial: bool = False, + http_status: int | None = None, + ) -> None: super().__init__(message) self.discard_partial = discard_partial + self.http_status = http_status @dataclass(frozen=True) @@ -162,6 +170,8 @@ class DownloadTelemetry: bytes_written: int = 0 write_seconds: float = 0.0 reason: str | None = None + http_status: int | None = None + resumed_bytes: int = 0 DownloadObserver = Callable[[DownloadTelemetry], None] @@ -827,6 +837,30 @@ class DepotSyncResult: completed_at: str files_planned: int = 0 bytes_planned: int = 0 + manifest_duration_ms: int = 0 + scan_duration_ms: int = 0 + download_duration_ms: int = 0 + apply_duration_ms: int = 0 + update_duration_ms: int = 0 + network_bytes_downloaded: int = 0 + download_retry_count: int = 0 + download_resume_used: bool = False + download_average_mbps: float | None = None + mirror_id: str | None = None + + +@dataclass(frozen=True) +class DepotDownloadMetrics: + """Aggregate network measurements for one depot download phase.""" + + duration_ms: int = 0 + transferred_bytes: int = 0 + retry_count: int = 0 + resume_used: bool = False + resumed_bytes: int = 0 + average_mbps: float | None = None + mirror_id: str | None = None + http_status: int | None = None @dataclass(frozen=True) @@ -959,6 +993,7 @@ class DepotClient: server=attempt_server, attempt=attempt_index + 1, reason=str(exc), + http_status=exc.http_status, ) ) if exc.discard_partial: @@ -1012,6 +1047,15 @@ class DepotClient: digest = hashlib.sha1(usedforsecurity=False) elif written == file.size: if digest.hexdigest().casefold() == file.sha1.casefold(): + self._notify_download( + DownloadTelemetry( + kind="resume", + file=file.path, + server=server, + attempt=attempt, + resumed_bytes=written, + ) + ) temporary.replace(destination) if progress is not None: progress(written) @@ -1020,6 +1064,15 @@ class DepotClient: written = 0 digest = hashlib.sha1(usedforsecurity=False) elif written and _recover_crlf_manifest_file(temporary, destination, file): + self._notify_download( + DownloadTelemetry( + kind="resume", + file=file.path, + server=server, + attempt=attempt, + resumed_bytes=written, + ) + ) temporary.unlink(missing_ok=True) if progress is not None: progress(file.size) @@ -1028,6 +1081,16 @@ class DepotClient: progress(written) requested_offset = written + if requested_offset: + self._notify_download( + DownloadTelemetry( + kind="resume", + file=file.path, + server=server, + attempt=attempt, + resumed_bytes=requested_offset, + ) + ) headers = _headers() if requested_offset: headers["Range"] = f"bytes={requested_offset}-" @@ -1051,14 +1114,17 @@ class DepotClient: raise _RetryableDownloadError( f"server rejected the saved offset for {file.path}", discard_partial=True, + http_status=status, ) from exc raise _RetryableDownloadError( - f"HTTP {status} from {server} for {file.path}" + f"HTTP {status} from {server} for {file.path}", + http_status=status, ) from exc except (OSError, HTTPException) as exc: raise _RetryableDownloadError( f"could not connect to {server} for {file.path}: {exc}" ) from exc + response_status = getattr(response, "status", None) self._notify_download( DownloadTelemetry( kind="response", @@ -1066,6 +1132,7 @@ class DepotClient: server=server, attempt=attempt, latency_seconds=max(0.0, time.monotonic() - request_started_at), + http_status=(response_status if isinstance(response_status, int) else None), ) ) @@ -1274,15 +1341,65 @@ def sync_managed_game( ) -> DepotSyncResult: """Synchronize the managed game directory using only official public data.""" + operation_started = time.perf_counter() + operation = ( + "repair" + if repair + else "update" + if paths.source_game.is_dir() + else "install" + ) + telemetry_enabled = not dry_run paths.create_base_directories() depot_client = client or DepotClient() previous_file_catalog = read_current_depot_file_catalog(paths) + current_version = _completed_depot_version(paths.native_depot_state) + manifest_started = time.perf_counter() + if operation == "update": + _send_depot_telemetry( + "update_check_started", + metadata={"current_version": current_version}, + context={"operation": operation, "stage": "manifest"}, + enabled=telemetry_enabled, + ) _emit_progress(progress, "manifest", 0, 0, 0, 0, None) - snapshot = depot_client.fetch_snapshot() + try: + snapshot = depot_client.fetch_snapshot() + except BaseException as exc: + _send_depot_telemetry( + "download_failed", + severity="error", + message=str(exc), + metadata={ + "resource_type": "manifest", + "duration_ms": _elapsed_milliseconds(manifest_started), + "failure_type": classify_network_failure(exc), + }, + context={"operation": operation, "stage": "manifest"}, + enabled=telemetry_enabled, + ) + raise + manifest_duration_ms = _elapsed_milliseconds(manifest_started) + scan_started = time.perf_counter() quarantine = QuarantineResult(0, None) - if quarantine_unmanaged and not dry_run: - quarantine = _quarantine_unmanaged_files(paths, snapshot) - plan = plan_update(paths.source_game, snapshot, repair=repair) + try: + if quarantine_unmanaged and not dry_run: + quarantine = _quarantine_unmanaged_files(paths, snapshot) + plan = plan_update(paths.source_game, snapshot, repair=repair) + except BaseException as exc: + _send_depot_telemetry( + "file_validation_failed", + severity="error", + message=str(exc), + metadata={ + "duration_ms": _elapsed_milliseconds(scan_started), + "repair": repair, + }, + context={"operation": operation, "stage": "scan"}, + enabled=telemetry_enabled, + ) + raise + scan_duration_ms = _elapsed_milliseconds(scan_started) _emit_progress( progress, "planned", @@ -1292,6 +1409,32 @@ def sync_managed_game( plan.bytes_total, None, ) + plan_metadata = { + "current_version": current_version, + "target_version": snapshot.version, + "changed_file_count": len(plan.downloads) + len(plan.deletes), + "files_checked": plan.files_total, + "files_missing": sum(item.reason == "missing" for item in plan.downloads), + "files_invalid": sum(item.reason != "missing" for item in plan.downloads), + "files_replaced": len(plan.downloads), + "download_bytes": plan.bytes_total, + "manifest_duration_ms": manifest_duration_ms, + "scan_duration_ms": scan_duration_ms, + } + if operation == "update": + _send_depot_telemetry( + "update_available" if plan.downloads or plan.deletes else "update_not_required", + metadata=plan_metadata, + context={"operation": operation, "stage": "scan"}, + enabled=telemetry_enabled, + ) + elif operation == "repair": + _send_depot_telemetry( + "repair_scan_completed", + metadata=plan_metadata, + context={"operation": operation, "stage": "scan"}, + enabled=telemetry_enabled, + ) if dry_run: return _sync_result( @@ -1303,43 +1446,96 @@ def sync_managed_game( repair, True, quarantine=quarantine, + manifest_duration_ms=manifest_duration_ms, + scan_duration_ms=scan_duration_ms, + update_duration_ms=_elapsed_milliseconds(operation_started), ) root = _prepare_game_root(paths.source_game, create=True) - _download_plan( + download_metrics = _download_plan( depot_client, snapshot, root, plan, progress, download_options or DownloadOptions(), + operation=operation, + telemetry_enabled=telemetry_enabled, ) - deleted = _apply_deletions(root, plan.deletes) - created = _apply_create_if_missing(root, snapshot.create_if_missing) - result = _sync_result( - snapshot, - plan, - len(plan.downloads), - plan.bytes_total, - deleted, - repair, - False, - created=created, - quarantine=quarantine, - ) - verified_paths = { - *plan.hash_verified_paths, - *(download.file.path for download in plan.downloads), - } - cache_depot_file_catalog( - paths.depot_file_catalog, - root, - snapshot, - verified_paths=verified_paths, - previous=previous_file_catalog, - ) - _write_json(paths.native_depot_state, asdict(result)) - cache_depot_launch_catalog(paths.depot_launch_catalog_cache, snapshot) + apply_started = time.perf_counter() + if operation == "update": + _send_depot_telemetry( + "update_apply_started", + metadata={ + "target_version": snapshot.version, + "changed_file_count": len(plan.downloads) + len(plan.deletes), + }, + context={"operation": operation, "stage": "apply"}, + enabled=telemetry_enabled, + ) + try: + deleted = _apply_deletions(root, plan.deletes) + created = _apply_create_if_missing(root, snapshot.create_if_missing) + verified_paths = { + *plan.hash_verified_paths, + *(download.file.path for download in plan.downloads), + } + cache_depot_file_catalog( + paths.depot_file_catalog, + root, + snapshot, + verified_paths=verified_paths, + previous=previous_file_catalog, + ) + cache_depot_launch_catalog(paths.depot_launch_catalog_cache, snapshot) + apply_duration_ms = _elapsed_milliseconds(apply_started) + update_duration_ms = _elapsed_milliseconds(operation_started) + result = _sync_result( + snapshot, + plan, + len(plan.downloads), + plan.bytes_total, + deleted, + repair, + False, + created=created, + quarantine=quarantine, + manifest_duration_ms=manifest_duration_ms, + scan_duration_ms=scan_duration_ms, + apply_duration_ms=apply_duration_ms, + update_duration_ms=update_duration_ms, + download_metrics=download_metrics, + ) + _write_json(paths.native_depot_state, asdict(result)) + except BaseException as exc: + _send_depot_telemetry( + "filesystem_error", + severity="error", + message=str(exc), + metadata={ + "duration_ms": _elapsed_milliseconds(apply_started), + "target_version": snapshot.version, + }, + context={"operation": operation, "stage": "apply"}, + enabled=telemetry_enabled, + ) + raise + if operation == "update": + _send_depot_telemetry( + "update_apply_completed", + metadata={ + **plan_metadata, + "apply_duration_ms": result.apply_duration_ms, + "download_duration_ms": result.download_duration_ms, + "update_duration_ms": result.update_duration_ms, + "network_bytes_downloaded": result.network_bytes_downloaded, + "average_download_mbps": result.download_average_mbps, + "retry_count": result.download_retry_count, + "resume_used": result.download_resume_used, + }, + context={"operation": operation, "stage": "apply"}, + enabled=telemetry_enabled, + ) _emit_progress( progress, "complete", @@ -1463,9 +1659,12 @@ def _download_plan( plan: UpdatePlan, progress: ProgressCallback | None, options: DownloadOptions, -) -> None: + *, + operation: str = "update", + telemetry_enabled: bool = True, +) -> DepotDownloadMetrics: if not plan.downloads: - return + return DepotDownloadMetrics(mirror_id=_server_display_name(depot_snapshot.server)) lock = threading.Lock() emit_lock = threading.Lock() @@ -1476,6 +1675,11 @@ def _download_plan( transferred_bytes = 0 retry_count = 0 retry_reason: str | None = None + resume_offsets: dict[str, int] = {} + mirrors: set[str] = set() + last_http_status: int | None = None + retry_events_sent = 0 + resume_events_sent = 0 active_workers = 0 target_workers = 0 last_progress_emitted_at = 0.0 @@ -1486,6 +1690,19 @@ def _download_plan( limiter = BandwidthLimiter(bandwidth_limit) if bandwidth_limit is not None else None previous_observer = client.download_observer previous_limiter = client.bandwidth_limiter + started_event, completed_event = _download_event_names(operation) + _send_depot_telemetry( + started_event, + metadata={ + "resource_type": "repair_file" if operation == "repair" else "game", + "download_bytes": plan.bytes_total, + "file_count": len(plan.downloads), + "mirror_id": _server_display_name(depot_snapshot.server), + "cdn_region": depot_snapshot.region, + }, + context={"operation": operation, "stage": "download"}, + enabled=telemetry_enabled, + ) def emit_download_progress( current_file: str, @@ -1543,7 +1760,8 @@ def _download_plan( progress(event) def observe_download(event: DownloadTelemetry) -> None: - nonlocal transferred_bytes, retry_count, retry_reason + nonlocal last_http_status, resume_events_sent, retry_count + nonlocal retry_events_sent, retry_reason, transferred_bytes if event.kind == "response" and event.latency_seconds is not None: controller.record_response(event.latency_seconds) elif event.kind == "retry": @@ -1552,9 +1770,12 @@ def _download_plan( controller.record_chunk(event.bytes_written, event.write_seconds) force = event.kind == "retry" - if event.kind in {"request", "response", "retry", "chunk"}: + emit_retry = False + emit_resume = False + if event.kind in {"request", "response", "retry", "resume", "chunk"}: with lock: current_servers[event.file] = _server_display_name(event.server) + mirrors.add(_server_display_name(event.server)) if event.kind == "chunk": transferred_bytes += event.bytes_written now = time.monotonic() @@ -1563,8 +1784,51 @@ def _download_plan( if event.kind == "retry": retry_count += 1 retry_reason = event.reason + if retry_events_sent < 5: + retry_events_sent += 1 + emit_retry = True + if event.kind == "resume" and event.resumed_bytes > 0: + prior_offset = resume_offsets.get(event.file, 0) + resume_offsets[event.file] = max(prior_offset, event.resumed_bytes) + if prior_offset == 0 and resume_events_sent < 5: + resume_events_sent += 1 + emit_resume = True + if event.http_status is not None: + last_http_status = event.http_status if previous_observer is not None: previous_observer(event) + if emit_retry: + _send_depot_telemetry( + "download_retry", + severity="warning", + message=event.reason, + metadata={ + "resource_type": ( + "repair_file" if operation == "repair" else "game" + ), + "file": event.file, + "attempt": event.attempt, + "failure_type": classify_network_failure(event.reason), + "http_status": event.http_status, + "mirror_id": _server_display_name(event.server), + }, + context={"operation": operation, "stage": "download_retry"}, + enabled=telemetry_enabled, + ) + if emit_resume: + _send_depot_telemetry( + "download_resumed", + metadata={ + "resource_type": ( + "repair_file" if operation == "repair" else "game" + ), + "file": event.file, + "resumed_bytes": event.resumed_bytes, + "mirror_id": _server_display_name(event.server), + }, + context={"operation": operation, "stage": "download_resume"}, + enabled=telemetry_enabled, + ) if event.kind != "chunk": emit_download_progress(event.file, force=force) @@ -1584,6 +1848,24 @@ def _download_plan( progress=report_file_bytes if progress is not None else None, ) + def metrics_snapshot() -> DepotDownloadMetrics: + with lock: + duration_seconds = max(0.0, time.monotonic() - started_at) + primary_mirror = _server_display_name(depot_snapshot.server) + return DepotDownloadMetrics( + duration_ms=round(duration_seconds * 1000), + transferred_bytes=transferred_bytes, + retry_count=retry_count, + resume_used=bool(resume_offsets), + resumed_bytes=sum(resume_offsets.values()), + average_mbps=_average_megabits_per_second( + transferred_bytes, + duration_seconds, + ), + mirror_id=primary_mirror, + http_status=last_http_status, + ) + maximum_workers = min(options.effective_maximum_workers, len(plan.downloads)) # Start large transfers first so they overlap the small-file request workload # instead of becoming a single-connection tail at the end of an install. @@ -1659,10 +1941,56 @@ def _download_plan( for future in future_items: future.cancel() raise + except BaseException as exc: + metrics = metrics_snapshot() + _send_depot_telemetry( + "download_failed", + severity="error", + message=str(exc), + metadata={ + "resource_type": "repair_file" if operation == "repair" else "game", + "download_bytes": metrics.transferred_bytes, + "content_bytes": plan.bytes_total, + "duration_ms": metrics.duration_ms, + "average_mbps": metrics.average_mbps, + "retry_count": metrics.retry_count, + "resume_used": metrics.resume_used, + "resumed_bytes": metrics.resumed_bytes, + "http_status": metrics.http_status, + "failure_type": classify_network_failure(exc), + "mirror_id": metrics.mirror_id, + "mirrors_tried": sorted(mirrors), + }, + context={"operation": operation, "stage": "download"}, + enabled=telemetry_enabled, + ) + raise finally: client.download_observer = previous_observer client.bandwidth_limiter = previous_limiter + metrics = metrics_snapshot() + _send_depot_telemetry( + completed_event, + metadata={ + "resource_type": "repair_file" if operation == "repair" else "game", + "download_bytes": metrics.transferred_bytes, + "content_bytes": plan.bytes_total, + "duration_ms": metrics.duration_ms, + "average_mbps": metrics.average_mbps, + "retry_count": metrics.retry_count, + "resume_used": metrics.resume_used, + "resumed_bytes": metrics.resumed_bytes, + "http_status": metrics.http_status, + "mirror_id": metrics.mirror_id, + "mirrors_used": sorted(mirrors), + "file_count": len(plan.downloads), + }, + context={"operation": operation, "stage": "download"}, + enabled=telemetry_enabled, + ) + return metrics + def _apply_deletions(root: Path, relative_paths: Iterable[str]) -> int: deleted = 0 @@ -1853,10 +2181,62 @@ def _completed_depot_build(path: Path) -> int | None: return build +def _completed_depot_version(path: Path) -> str | None: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + if not isinstance(value, dict) or value.get("dry_run") is not False: + return None + version = value.get("version") + return version[:32] if isinstance(version, str) and version else None + + def _elapsed_milliseconds(started: float) -> int: return max(0, round((time.perf_counter() - started) * 1000)) +def _average_megabits_per_second(byte_count: int, duration_seconds: float) -> float | None: + if byte_count <= 0 or duration_seconds <= 0: + return None + return round((byte_count * 8) / duration_seconds / 1_000_000, 3) + + +def _download_event_names(operation: str) -> tuple[str, str]: + if operation == "install": + return "install_download_started", "install_download_completed" + if operation == "update": + return "update_download_started", "update_download_completed" + return "download_started", "download_completed" + + +def _send_depot_telemetry( + event: str, + *, + severity: str = "info", + message: str | None = None, + metadata: Mapping[str, object] | None = None, + context: Mapping[str, object] | None = None, + enabled: bool = True, +) -> None: + if not enabled: + return + telemetry = current_telemetry() + if telemetry is None: + return + try: + telemetry.send_event( + event, + severity=severity, + message=message, + metadata=metadata, + context=context, + ) + except Exception: + # Telemetry must never alter verified depot behavior. + pass + + def _sync_result( snapshot: DepotSnapshot, plan: UpdatePlan, @@ -1868,7 +2248,13 @@ def _sync_result( *, created: int = 0, quarantine: QuarantineResult = QuarantineResult(0, None), + manifest_duration_ms: int = 0, + scan_duration_ms: int = 0, + apply_duration_ms: int = 0, + update_duration_ms: int = 0, + download_metrics: DepotDownloadMetrics | None = None, ) -> DepotSyncResult: + metrics = download_metrics or DepotDownloadMetrics() return DepotSyncResult( region=snapshot.region, server=snapshot.server, @@ -1887,6 +2273,16 @@ def _sync_result( repair=repair, dry_run=dry_run, completed_at=datetime.now().astimezone().isoformat(), + manifest_duration_ms=manifest_duration_ms, + scan_duration_ms=scan_duration_ms, + download_duration_ms=metrics.duration_ms, + apply_duration_ms=apply_duration_ms, + update_duration_ms=update_duration_ms, + network_bytes_downloaded=metrics.transferred_bytes, + download_retry_count=metrics.retry_count, + download_resume_used=metrics.resume_used, + download_average_mbps=metrics.average_mbps, + mirror_id=metrics.mirror_id, ) diff --git a/ptlinux/importer.py b/ptlinux/importer.py index 7a8c4a5..245a3d9 100644 --- a/ptlinux/importer.py +++ b/ptlinux/importer.py @@ -101,6 +101,7 @@ def import_existing_game( runner.run( ["cp", "-a", "--reflink=auto", "--", f"{source}/.", str(temporary)], timeout_seconds=1800.0, + telemetry_stage="copy", ) source_after = inventory_game_tree(source) copied = inventory_game_tree(temporary) diff --git a/ptlinux/launcher.py b/ptlinux/launcher.py index 7d49f01..82fe302 100644 --- a/ptlinux/launcher.py +++ b/ptlinux/launcher.py @@ -189,6 +189,7 @@ def configure_launcher_prefix( [str(tools.wineboot), "-u"], env_overrides={**environment, "WINEARCH": "win64"}, timeout_seconds=300.0, + telemetry_stage="prefix_create", ) ] if force_runtime or not _dotnet_desktop_ready(paths, marker): @@ -197,6 +198,7 @@ def configure_launcher_prefix( [str(tools.winetricks), "-q", "dotnetdesktop6"], env_overrides=environment, timeout_seconds=1800.0, + telemetry_stage="runtime_install", ) ) @@ -214,7 +216,12 @@ def configure_launcher_prefix( ) for command in registry_commands: results.append( - runner.run(command, env_overrides=environment, timeout_seconds=60.0) + runner.run( + command, + env_overrides=environment, + timeout_seconds=60.0, + telemetry_stage="prefix_configuration", + ) ) for requested, replacement in ( ("Segoe UI", "Carlito"), @@ -236,6 +243,7 @@ def configure_launcher_prefix( ], env_overrides=environment, timeout_seconds=60.0, + telemetry_stage="font_configuration", ) ) if not runner.dry_run: @@ -263,6 +271,7 @@ def launch_official_updater( cwd=paths.launcher_patched.parent, env_overrides=launcher_environment(paths, tools), timeout_seconds=4 * 60 * 60, + telemetry_stage="launcher_start", ) diff --git a/ptlinux/paths.py b/ptlinux/paths.py index a2f485a..4899c33 100644 --- a/ptlinux/paths.py +++ b/ptlinux/paths.py @@ -279,6 +279,12 @@ class AppPaths: def settings_lock(self) -> Path: return self.config / "settings.lock" + @property + def installation_id_file(self) -> Path: + """Anonymous telemetry identity stored separately from user preferences.""" + + return self.config / "install-id" + def create_base_directories(self) -> None: for path in ( self.data, diff --git a/ptlinux/runtime.py b/ptlinux/runtime.py index b434d03..fabf73a 100644 --- a/ptlinux/runtime.py +++ b/ptlinux/runtime.py @@ -273,6 +273,7 @@ def prepare_game_runtime( [selection.executable, "winetricks", "-q", "d3dx9_43", "d3dcompiler_43"], env_overrides=environment, timeout_seconds=1800.0, + telemetry_stage="directx_install", ) ) elif selection.kind in {"proton7", "proton7-container"}: @@ -299,6 +300,7 @@ def prepare_game_runtime( ], env_overrides=environment, timeout_seconds=600.0, + telemetry_stage="prefix_create", ) ) if not runner.dry_run: @@ -324,6 +326,7 @@ def prepare_game_runtime( ], env_overrides=environment, timeout_seconds=60.0, + telemetry_stage="dxvk_configuration", ) ) if not native_d3d_ready(paths.proton_prefix): @@ -352,6 +355,7 @@ def prepare_game_runtime( "WINEDEBUG": "-all", }, timeout_seconds=1800.0, + telemetry_stage="directx_install", ) ) else: @@ -504,6 +508,7 @@ def run_game( stream_output=True, output_line_filter=output_line_filter, pass_fds=pass_fds, + telemetry_stage="game_start", ) @@ -599,6 +604,7 @@ def run_runtime_prefix_command( env_overrides=environment, timeout_seconds=timeout_seconds, check=check, + telemetry_stage="runtime_check", ) diff --git a/ptlinux/self_update.py b/ptlinux/self_update.py index 77f27f8..c18fe94 100644 --- a/ptlinux/self_update.py +++ b/ptlinux/self_update.py @@ -15,6 +15,7 @@ import subprocess import sys import tarfile import tempfile +import time from typing import Mapping from urllib.error import HTTPError, URLError from urllib.parse import urlsplit, urlunsplit @@ -23,6 +24,11 @@ from urllib.request import Request, urlopen from . import __version__ from .locking import launcher_operation_lock from .paths import AppPaths +from .telemetry import ( + classify_network_failure, + current_telemetry, + network_http_status, +) DEFAULT_UPDATE_REPOSITORY_URL = ( @@ -681,6 +687,18 @@ def _download_archive( ) -> None: if destination.is_file() and _archive_matches(destination, manifest): return + telemetry = current_telemetry() + download_started_at = time.monotonic() + if telemetry is not None: + telemetry.send_event( + "download_started", + metadata={ + "resource_type": "launcher_update", + "content_bytes": manifest.archive_size, + "mirror_id": "launcher_update_channel", + }, + context={"operation": "launcher_update", "stage": "download"}, + ) destination.parent.mkdir(parents=True, exist_ok=True) temporary = destination.with_name(f".{destination.name}.{os.getpid()}.part") temporary.unlink(missing_ok=True) @@ -693,6 +711,7 @@ def _download_archive( }, method="GET", ) + written = 0 try: try: response = urlopen(request, timeout=timeout_seconds) @@ -715,7 +734,6 @@ def _download_archive( "launcher archive Content-Length does not match the signed manifest" ) digest = hashlib.sha256() - written = 0 with temporary.open("xb") as handle: while True: chunk = response.read(READ_SIZE) @@ -736,9 +754,52 @@ def _download_archive( raise LauncherUpdateError("launcher archive SHA-256 does not match the manifest") temporary.replace(destination) _fsync_directory(destination.parent) - except BaseException: + except BaseException as exc: temporary.unlink(missing_ok=True) + if telemetry is not None: + duration = max(0.0, time.monotonic() - download_started_at) + telemetry.send_event( + "download_failed", + severity="error", + message=str(exc), + metadata={ + "resource_type": "launcher_update", + "download_bytes": written, + "content_bytes": manifest.archive_size, + "duration_ms": round(duration * 1000), + "average_mbps": _average_mbps(written, duration), + "retry_count": 0, + "resume_used": False, + "http_status": network_http_status(exc), + "failure_type": classify_network_failure(exc), + "mirror_id": "launcher_update_channel", + }, + context={"operation": "launcher_update", "stage": "download"}, + ) raise + duration = max(0.0, time.monotonic() - download_started_at) + if telemetry is not None: + telemetry.send_event( + "download_completed", + metadata={ + "resource_type": "launcher_update", + "download_bytes": written, + "content_bytes": manifest.archive_size, + "duration_ms": round(duration * 1000), + "average_mbps": _average_mbps(written, duration), + "retry_count": 0, + "resume_used": False, + "http_status": 200, + "mirror_id": "launcher_update_channel", + }, + context={"operation": "launcher_update", "stage": "download"}, + ) + + +def _average_mbps(download_bytes: int, duration_seconds: float) -> float | None: + if download_bytes <= 0 or duration_seconds <= 0: + return None + return round((download_bytes * 8) / duration_seconds / 1_000_000, 3) def _stage_and_install_release( diff --git a/ptlinux/settings.py b/ptlinux/settings.py index 8d0b958..3ded4db 100644 --- a/ptlinux/settings.py +++ b/ptlinux/settings.py @@ -37,6 +37,7 @@ _RETIRED_SETTINGS = frozenset( class Settings: realm: str = "Babel" ui_scale: str = "auto" + telemetry_enabled: bool = True borderless: bool = False suppress_unresponsive_warnings: bool = True runtime: str = "auto" @@ -154,6 +155,7 @@ def _settings_from_object(value: dict[str, object]) -> Settings: settings = Settings( realm=_string(value, "realm", "Babel").strip(), ui_scale=_string(value, "ui_scale", "auto"), + telemetry_enabled=_boolean(value, "telemetry_enabled", True), borderless=_migrated_borderless(value), allow_window_resizing=_boolean( value, diff --git a/ptlinux/telemetry.py b/ptlinux/telemetry.py new file mode 100644 index 0000000..50a428c --- /dev/null +++ b/ptlinux/telemetry.py @@ -0,0 +1,1541 @@ +"""Anonymous, best-effort launcher telemetry with bounded asynchronous delivery.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +import hashlib +import json +import math +import os +from pathlib import Path +import platform +import queue +import re +import resource +import socket +import sys +import threading +import time +import traceback +from types import TracebackType +from typing import Any +from urllib.parse import parse_qsl, urlsplit, urlunsplit +from urllib.request import Request, urlopen +import uuid + +from . import __version__ +from .paths import AppPaths +from .telemetry_environment import collect_extended_environment + + +DEFAULT_TELEMETRY_ENDPOINT = "https://pristontale.eu/api/launcher/telemetry" +TELEMETRY_ENDPOINT_VARIABLE = "PTLINUX_TELEMETRY_ENDPOINT" +TELEMETRY_KEY_VARIABLE = "PTLINUX_TELEMETRY_KEY" +TELEMETRY_DEBUG_VARIABLE = "PTLINUX_TELEMETRY_DEBUG" + +MAX_REQUEST_BYTES = 64 * 1024 +MAX_MESSAGE_BYTES = 10 * 1024 +MAX_CONTEXT_BYTES = 16 * 1024 +MAX_METADATA_BYTES = 32 * 1024 +MAX_STDERR_CHARACTERS = 4000 +MAX_STDOUT_CHARACTERS = 2000 +MAX_TRACEBACK_CHARACTERS = 10_000 +MAX_CONTAINER_ITEMS = 128 +MAX_VALUE_DEPTH = 8 +QUEUE_CAPACITY = 128 +DELIVERY_TIMEOUT_SECONDS = 3.0 +SHUTDOWN_WAIT_SECONDS = 0.25 +ERROR_SUPPRESSION_WINDOW_SECONDS = 60.0 +MAX_ERROR_FINGERPRINTS = 128 + +_RESOURCE_USAGE_EVENTS = frozenset( + { + "dependency_fix_success", + "dxvk_install_success", + "dxvk_repair_success", + "game_exited", + "import_success", + "install_success", + "launcher_closed", + "launcher_ready", + "prefix_reset_success", + "repair_success", + "runtime_install_success", + "runtime_repair_success", + "update_success", + } +) + +VALID_SEVERITIES = frozenset({"debug", "info", "warning", "error", "critical"}) +VALID_EVENTS = frozenset( + { + "api_error", + "asyncio_exception", + "checksum_failed", + "compatibility_check_completed", + "compatibility_check_failed", + "compatibility_check_started", + "configuration_error", + "dependency_fix_failed", + "dependency_fix_requested", + "dependency_fix_started", + "dependency_fix_success", + "dependency_check_failed", + "download_completed", + "download_failed", + "download_resumed", + "download_retry", + "download_started", + "dxvk_install_failed", + "dxvk_install_started", + "dxvk_install_success", + "dxvk_repair_failed", + "dxvk_repair_started", + "dxvk_repair_success", + "file_validation_failed", + "filesystem_error", + "first_run_completed", + "first_run_failed", + "first_run_started", + "game_exited", + "game_launch_failed", + "game_launch_requested", + "game_launch_started", + "game_launch_success", + "import_failed", + "import_scan_completed", + "import_started", + "import_success", + "install_download_completed", + "install_download_started", + "install_extract_completed", + "install_extract_started", + "install_failed", + "install_started", + "install_success", + "launcher_closed", + "launcher_exception", + "launcher_ready", + "launcher_started", + "operation_timeout", + "permission_error", + "prefix_reset_failed", + "prefix_reset_requested", + "prefix_reset_started", + "prefix_reset_success", + "repair_failed", + "repair_requested", + "repair_scan_completed", + "repair_started", + "repair_success", + "repeated_error", + "runtime_check_failed", + "runtime_install_failed", + "runtime_install_started", + "runtime_install_success", + "runtime_reinstall_requested", + "runtime_repair_failed", + "runtime_repair_started", + "runtime_repair_success", + "subprocess_failed", + "thread_exception", + "update_apply_completed", + "update_apply_started", + "update_available", + "update_check_started", + "update_download_completed", + "update_download_started", + "update_failed", + "update_not_required", + "update_started", + "update_success", + "vulkan_32bit_check_failed", + "vulkan_check_failed", + "wine_32bit_check_failed", + } +) + +_SECRET_KEYS = frozenset( + { + "api_key", + "apikey", + "authorization", + "client_secret", + "cookie", + "launcher_key", + "password", + "passwd", + "proxy_authorization", + "refresh_token", + "secret", + "session_id", + "sessionid", + "set_cookie", + "token", + "access_token", + } +) +_SECRET_QUERY_KEYS = _SECRET_KEYS | frozenset( + { + "expires", + "key-pair-id", + "policy", + "signature", + "sig", + "x-amz-credential", + "x-amz-signature", + "x-amz-security-token", + } +) +_NETWORK_FILESYSTEMS = frozenset( + {"9p", "afs", "ceph", "cifs", "nfs", "nfs4", "smb3", "sshfs"} +) +_URL_PATTERN = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE) +_EMAIL_PATTERN = re.compile( + r"(?|/home/|/root)[\\/]\.ssh(?:[\\/]\S*)?") +_AUTHORIZATION_PATTERN = re.compile( + r"(?i)(authorization\s*:\s*(?:bearer|basic)\s+)[^\s,;]+" +) +_SECRET_ASSIGNMENT_PATTERN = re.compile( + r"(?i)\b(password|passwd|token|access_token|refresh_token|api[_-]?key|" + r"client_secret|session_id|cookie)\s*([=:]\s*)(?:\"[^\"]*\"|'[^']*'|[^&\s,;]+)" +) +_UUID_PATTERN = re.compile( + r"\b[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}\b", + re.IGNORECASE, +) +_MEMORY_ADDRESS_PATTERN = re.compile(r"\b0x[0-9a-f]{6,16}\b", re.IGNORECASE) +_HTTP_STATUS_PATTERN = re.compile(r"\bHTTP\s+([1-5][0-9]{2})\b", re.IGNORECASE) +_INSTALL_ID_FILE_MODE = 0o600 + +TelemetryTransport = Callable[[str, bytes, Mapping[str, str], float], None] + + +@dataclass +class _RepeatedErrorState: + fingerprint: str + event: str + first_seen: float + last_seen: float + suppressed_count: int + operation: str | None + stage: str | None + + +class TelemetryService: + """Own telemetry collection, sanitization, queuing, and delivery. + + Every public method is best-effort. Callers must never need error handling to + protect a launcher operation from telemetry failures. + """ + + def __init__( + self, + paths: AppPaths, + *, + enabled: bool = True, + endpoint: str | None = None, + launcher_key: str | None = None, + environment: Mapping[str, str] | None = None, + transport: TelemetryTransport | None = None, + session_metadata: Mapping[str, object] | None = None, + monotonic: Callable[[], float] = time.monotonic, + ) -> None: + self.paths = paths + self._environment = os.environ if environment is None else environment + configured_endpoint = self._environment.get(TELEMETRY_ENDPOINT_VARIABLE) + self.endpoint = ( + configured_endpoint.strip() + if configured_endpoint is not None + else (endpoint or DEFAULT_TELEMETRY_ENDPOINT) + ) + configured_key = self._environment.get(TELEMETRY_KEY_VARIABLE) + self._launcher_key = launcher_key if launcher_key is not None else configured_key + self._transport = transport or _post_json + self._session_metadata = dict(session_metadata or {}) + self._monotonic = monotonic + self._started_at = monotonic() + self._enabled = bool(enabled) + self._active = False + self._accepting = False + self._ready = False + self._first_run = False + self._installation_id: str | None = None + self._base_payload: dict[str, object] = {} + self._queue: queue.Queue[bytes] = queue.Queue(maxsize=QUEUE_CAPACITY) + self._stop = threading.Event() + self._worker: threading.Thread | None = None + self._environment_worker: threading.Thread | None = None + self._state_lock = threading.RLock() + self._repeated_errors: dict[str, _RepeatedErrorState] = {} + self._previous_sys_excepthook: Any = None + self._previous_threading_excepthook: Any = None + self._installed_sys_excepthook: Any = None + self._installed_threading_excepthook: Any = None + self._debug = self._environment.get(TELEMETRY_DEBUG_VARIABLE, "").casefold() in { + "1", + "true", + "yes", + } + private_values = { + self._environment.get("USER", ""), + self._environment.get("LOGNAME", ""), + socket.gethostname(), + } + self._private_literals = tuple( + sorted( + (value for value in private_values if len(value) >= 3), + key=len, + reverse=True, + ) + ) + + @property + def enabled(self) -> bool: + return self._enabled and self._active + + @property + def installation_id(self) -> str | None: + return self._installation_id + + def initialize(self) -> None: + """Initialize the service once without propagating telemetry errors.""" + + with self._state_lock: + if self._active or not self._enabled: + return + try: + _validate_endpoint(self.endpoint) + identity, created, first_seen = _load_or_create_installation_id( + self.paths.installation_id_file + ) + self._installation_id = identity + self._first_run = created + self._base_payload = self._collect_common_payload(first_seen) + self._stop.clear() + self._accepting = True + self._active = True + worker = threading.Thread( + target=self._worker_loop, + name="ptlinux-telemetry", + daemon=True, + ) + self._worker = worker + worker.start() + self._install_exception_hooks() + except Exception as exc: + self._enabled = False + self._accepting = False + self._active = False + self._debug_log(f"initialization failed: {type(exc).__name__}") + return + + self.send_event( + "launcher_started", + metadata={"cold_start": True, **self._session_metadata}, + ) + if self._first_run: + self.send_event( + "first_run_started", + metadata={"install_type": self._install_type()}, + ) + self._start_environment_collection() + + def set_enabled(self, enabled: bool) -> None: + """Apply a consent change to the current process.""" + + if enabled: + with self._state_lock: + if self._enabled and self._active: + return + if self._worker is not None and self._worker.is_alive(): + return + self._enabled = True + self.initialize() + return + self._deactivate() + + def mark_ready(self, metadata: Mapping[str, object] | None = None) -> None: + with self._state_lock: + if self._ready or not self._active: + return + self._ready = True + duration_ms = round((self._monotonic() - self._started_at) * 1000) + self.send_event( + "launcher_ready", + metadata={"startup_duration_ms": duration_ms, **dict(metadata or {})}, + ) + if self._first_run: + self.send_event( + "first_run_completed", + metadata={ + "duration_ms": duration_ms, + "install_type": self._install_type(), + }, + ) + + def send_event( + self, + event: str, + severity: str = "info", + message: str | None = None, + metadata: Mapping[str, object] | None = None, + context: Mapping[str, object] | None = None, + ) -> bool: + """Sanitize and enqueue one event without blocking on network I/O.""" + + try: + if ( + not self._accepting + or not self._active + or event not in VALID_EVENTS + or severity not in VALID_SEVERITIES + or self._installation_id is None + ): + return False + suppressed, summaries, repeated_error = self._repeated_error_decision( + event, + severity, + message, + metadata, + context, + ) + for summary in summaries: + self._send_repeated_error(summary) + if suppressed: + return True + with self._state_lock: + base_payload = dict(self._base_payload) + payload: dict[str, object] = { + "installation_id": self._installation_id, + "event": event, + "severity": severity, + "client_timestamp": datetime.now(timezone.utc).isoformat(), + **base_payload, + } + if message is not None: + payload["message"] = _truncate_utf8( + self._sanitize_text(str(message)), + MAX_MESSAGE_BYTES, + ) + if context: + sanitized_context = self._sanitize_value(dict(context), key=None, depth=0) + if isinstance(sanitized_context, dict): + payload["context"] = _bound_mapping( + sanitized_context, + MAX_CONTEXT_BYTES, + ) + event_metadata = dict(metadata or {}) + if event in _RESOURCE_USAGE_EVENTS: + for key, value in _resource_usage().items(): + event_metadata.setdefault(key, value) + if event_metadata: + sanitized_metadata = self._sanitize_value( + event_metadata, + key=None, + depth=0, + ) + if isinstance(sanitized_metadata, dict): + payload["metadata"] = _bound_mapping( + sanitized_metadata, + MAX_METADATA_BYTES, + ) + body = _encode_payload(payload) + if body is None: + return False + self._queue.put_nowait(body) + if repeated_error is not None: + evicted = self._remember_repeated_error(repeated_error) + if evicted is not None: + self._send_repeated_error(evicted) + self._debug_log(f"enqueued {event}") + return True + except (OSError, TypeError, ValueError, OverflowError, queue.Full): + return False + + def report_exception( + self, + event: str, + exception: BaseException, + metadata: Mapping[str, object] | None = None, + context: Mapping[str, object] | None = None, + *, + severity: str = "error", + ) -> bool: + try: + frames = traceback.extract_tb(exception.__traceback__) + last_frame = frames[-1] if frames else None + formatted = "".join( + traceback.TracebackException.from_exception(exception).format() + ) + details: dict[str, object] = dict(metadata or {}) + details.update( + { + "exception_type": type(exception).__name__, + "exception_module": type(exception).__module__, + "sanitized_traceback": _truncate_characters( + formatted, + MAX_TRACEBACK_CHARACTERS, + tail=True, + ), + } + ) + if last_frame is not None: + details["source_module"] = Path(last_frame.filename).name + details["source_function"] = last_frame.name + return self.send_event( + event, + severity=severity, + message=str(exception), + metadata=details, + context=context, + ) + except Exception: + return False + + def report_subprocess_failure( + self, + event: str, + result: object, + command: Sequence[object] | None = None, + metadata: Mapping[str, object] | None = None, + ) -> bool: + try: + returncode = getattr(result, "returncode", None) + timed_out = bool(getattr(result, "timed_out", False)) + if returncode == 0 and not timed_out: + return False + details: dict[str, object] = dict(metadata or {}) + details.update( + { + "exit_code": returncode, + "timed_out": timed_out, + "duration_ms": round( + float(getattr(result, "duration_seconds", 0.0)) * 1000 + ), + "stdout": _truncate_characters( + str(getattr(result, "stdout", "")), + MAX_STDOUT_CHARACTERS, + tail=True, + ), + "stderr": _truncate_characters( + str(getattr(result, "stderr", "")), + MAX_STDERR_CHARACTERS, + tail=True, + ), + } + ) + arguments = command + if arguments is None: + candidate = getattr(result, "argv", None) + if isinstance(candidate, Sequence) and not isinstance(candidate, str): + arguments = candidate + if arguments: + details["command"] = [str(value) for value in arguments] + if isinstance(returncode, int) and returncode < 0: + details["signal"] = -returncode + operation = details.get("operation") + stage = details.get("stage") + failure_context = { + "operation": operation or "subprocess", + "stage": stage if isinstance(stage, str) else "execution", + } + status = "unknown" if returncode is None else str(returncode) + return self.send_event( + event, + severity="error", + message=( + "Subprocess timed out" + if timed_out + else f"Subprocess exited with status {status}" + ), + metadata=details, + context=failure_context, + ) + except Exception: + return False + + def _repeated_error_decision( + self, + event: str, + severity: str, + message: str | None, + metadata: Mapping[str, object] | None, + context: Mapping[str, object] | None, + ) -> tuple[ + bool, + tuple[_RepeatedErrorState, ...], + _RepeatedErrorState | None, + ]: + if event == "repeated_error" or not _is_error_event(event, severity): + return False, (), None + details = metadata or {} + sanitized_message = _truncate_utf8( + self._sanitize_text(str(message or "")), + MAX_MESSAGE_BYTES, + ) + fingerprint = error_fingerprint( + event, + sanitized_message, + exception_type=self._sanitize_text( + _string_field(details, "exception_type") + ), + source_module=self._sanitize_text( + _string_field(details, "source_module") + ), + source_function=self._sanitize_text( + _string_field(details, "source_function") + ), + ) + now = self._monotonic() + summaries: list[_RepeatedErrorState] = [] + with self._state_lock: + current = self._repeated_errors.get(fingerprint) + if ( + current is not None + and now - current.last_seen <= ERROR_SUPPRESSION_WINDOW_SECONDS + ): + current.last_seen = now + current.suppressed_count += 1 + return True, (), None + if current is not None: + self._repeated_errors.pop(fingerprint, None) + if current.suppressed_count > 0: + summaries.append(current) + operation = _string_field(context or {}, "operation") or None + stage = _string_field(context or {}, "stage") or None + candidate = _RepeatedErrorState( + fingerprint=fingerprint, + event=event, + first_seen=now, + last_seen=now, + suppressed_count=0, + operation=operation, + stage=stage, + ) + return False, tuple(summaries), candidate + + def _remember_repeated_error( + self, + state: _RepeatedErrorState, + ) -> _RepeatedErrorState | None: + with self._state_lock: + if state.fingerprint in self._repeated_errors: + return None + self._repeated_errors[state.fingerprint] = state + if len(self._repeated_errors) <= MAX_ERROR_FINGERPRINTS: + return None + oldest_fingerprint, oldest = min( + self._repeated_errors.items(), + key=lambda item: item[1].last_seen, + ) + self._repeated_errors.pop(oldest_fingerprint, None) + return oldest if oldest.suppressed_count > 0 else None + + def _send_repeated_error(self, state: _RepeatedErrorState) -> None: + metadata: dict[str, object] = { + "fingerprint": state.fingerprint, + "original_event": state.event, + "suppressed_count": state.suppressed_count, + "window_seconds": round( + max(0.0, state.last_seen - state.first_seen), + 3, + ), + } + if state.stage: + metadata["original_stage"] = state.stage + self.send_event( + "repeated_error", + severity="warning", + metadata=metadata, + context={ + "operation": state.operation or "telemetry", + "stage": "error_suppression", + }, + ) + + def _flush_repeated_errors(self) -> None: + with self._state_lock: + summaries = tuple( + state + for state in self._repeated_errors.values() + if state.suppressed_count > 0 + ) + self._repeated_errors.clear() + for state in summaries: + self._send_repeated_error(state) + + def shutdown(self, timeout_seconds: float = SHUTDOWN_WAIT_SECONDS) -> None: + """Queue the closing event and give the daemon worker a bounded flush.""" + + with self._state_lock: + if not self._active: + return + deadline = time.monotonic() + max(0.0, timeout_seconds) + environment_worker = self._environment_worker + if isinstance(environment_worker, threading.Thread) and ( + environment_worker is not threading.current_thread() + ): + environment_worker.join( + timeout=min( + 0.5, + max(0.0, (deadline - time.monotonic()) / 2), + ) + ) + self._flush_repeated_errors() + runtime_seconds = max(0, round(self._monotonic() - self._started_at)) + if self._first_run and not self._ready: + self.send_event( + "first_run_failed", + severity="error", + message="Launcher did not reach the ready state", + metadata={"duration_ms": runtime_seconds * 1000}, + ) + self.send_event( + "launcher_closed", + metadata={"runtime_seconds": runtime_seconds, **self._session_metadata}, + ) + with self._state_lock: + self._accepting = False + self._stop.set() + worker = self._worker + if isinstance(worker, threading.Thread) and ( + worker is not threading.current_thread() + ): + worker.join(timeout=max(0.0, deadline - time.monotonic())) + self._restore_exception_hooks() + with self._state_lock: + self._active = False + + def _deactivate(self) -> None: + with self._state_lock: + self._enabled = False + self._accepting = False + self._stop.set() + self._repeated_errors.clear() + while True: + try: + self._queue.get_nowait() + except queue.Empty: + break + else: + self._queue.task_done() + self._restore_exception_hooks() + with self._state_lock: + self._active = False + + def _worker_loop(self) -> None: + headers = { + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)", + } + if self._launcher_key: + headers["X-Launcher-Key"] = self._launcher_key + while True: + try: + body = self._queue.get(timeout=0.1) + except queue.Empty: + if self._stop.is_set(): + return + continue + try: + if self._enabled: + self._transport( + self.endpoint, + body, + headers, + DELIVERY_TIMEOUT_SECONDS, + ) + self._debug_log("send succeeded") + except Exception as exc: + self._debug_log(f"send failed: {type(exc).__name__}") + finally: + self._queue.task_done() + + def _start_environment_collection(self) -> None: + with self._state_lock: + if not self._active or not self._enabled: + return + worker = threading.Thread( + target=self._collect_extended_environment, + name="ptlinux-telemetry-environment", + daemon=True, + ) + self._environment_worker = worker + worker.start() + + def _collect_extended_environment(self) -> None: + started_at = self._monotonic() + try: + details = collect_extended_environment(self.paths, self._environment) + sanitized = self._sanitize_value(details, key=None, depth=0) + if not isinstance(sanitized, dict): + return + except Exception as exc: + self._debug_log( + f"extended environment detection failed: {type(exc).__name__}" + ) + return + + with self._state_lock: + if not self._active or not self._enabled: + return + for key, value in sanitized.items(): + existing = self._base_payload.get(key) + if isinstance(existing, dict) and isinstance(value, dict): + self._base_payload[key] = {**existing, **value} + else: + self._base_payload[key] = value + + runtime = sanitized.get("runtime") + dependencies = sanitized.get("dependencies") + runtime_details = runtime if isinstance(runtime, dict) else {} + dependency_details = dependencies if isinstance(dependencies, dict) else {} + metadata = { + "automatic": True, + "duration_ms": round((self._monotonic() - started_at) * 1000), + "dependency_healthy": dependency_details.get("healthy"), + "missing_required": dependency_details.get("missing_required", []), + "runtime_kind": runtime_details.get("kind"), + "dxvk_install_status": runtime_details.get("dxvk_install_status"), + "vulkan_64bit": runtime_details.get("vulkan_64bit"), + "vulkan_32bit": runtime_details.get("vulkan_32bit"), + } + self.send_event( + "compatibility_check_completed", + metadata=metadata, + context={"operation": "launcher", "stage": "environment_detection"}, + ) + missing_required = dependency_details.get("missing_required") + if dependency_details.get("healthy") is False: + self.send_event( + "dependency_check_failed", + severity="warning", + message="One or more launcher dependencies are unavailable", + metadata={ + "dependencies": ( + missing_required if isinstance(missing_required, list) else [] + ), + }, + context={"operation": "launcher", "stage": "dependency_check"}, + ) + if dependency_details.get("wine_32bit_available") is False: + self.send_event( + "wine_32bit_check_failed", + severity="warning", + metadata={"dependency": "wine_32bit", "check": "32bit_wine"}, + context={"operation": "launcher", "stage": "runtime_check"}, + ) + if runtime_details.get("vulkan_64bit") is False: + self.send_event( + "vulkan_check_failed", + severity="warning", + metadata={"dependency": "vulkan", "check": "64bit_vulkan"}, + context={"operation": "launcher", "stage": "vulkan_check"}, + ) + if runtime_details.get("vulkan_32bit") is False: + self.send_event( + "vulkan_32bit_check_failed", + severity="warning", + metadata={"dependency": "vulkan_32bit", "check": "32bit_vulkan"}, + context={"operation": "launcher", "stage": "vulkan_check"}, + ) + + def _collect_common_payload(self, first_seen: str) -> dict[str, object]: + environment = self._environment + payload: dict[str, object] = { + "launcher_version": __version__, + "launcher_build": environment.get("PTLINUX_BUILD", __version__)[:32], + "launcher_channel": environment.get("PTLINUX_CHANNEL", "stable")[:16], + "python_version": platform.python_version(), + "installation": { + "first_seen": first_seen, + "install_type": self._install_type(), + "install_is_external": _install_is_external(self.paths, environment), + "configuration_schema_version": _settings_schema_version(), + }, + "os": _operating_system_info(), + "display": _display_info(environment), + "filesystem": _filesystem_info(self.paths.data), + } + commit = environment.get("PTLINUX_COMMIT") + if commit and re.fullmatch(r"[0-9a-fA-F]{7,40}", commit): + payload["launcher_commit"] = commit[:12].lower() + game_version = _current_game_version(self.paths.native_depot_state) + if game_version is not None: + payload["game_version"] = game_version + filesystem = payload["filesystem"] + if isinstance(filesystem, dict): + installation = payload["installation"] + if isinstance(installation, dict): + installation["install_is_network_fs"] = bool( + filesystem.get("network") + ) + return self._sanitize_value(payload, key=None, depth=0) # type: ignore[return-value] + + def _install_type(self) -> str: + if self.paths.game_import_manifest.is_file(): + return "imported" + if self.paths.source_game.is_dir(): + return "fresh" + return "unknown" + + def _sanitize_value(self, value: object, *, key: str | None, depth: int) -> object: + if key is not None and _is_secret_key(key): + return "" + if depth >= MAX_VALUE_DEPTH: + return "" + if value is None or isinstance(value, (bool, int)): + return value + if isinstance(value, float): + return value if math.isfinite(value) else None + if isinstance(value, (str, Path)): + text = self._sanitize_text(str(value)) + limit = _string_limit_for_key(key) + return _truncate_characters(text, limit, tail=_keep_string_tail(key)) + if isinstance(value, Mapping): + result: dict[str, object] = {} + for index, (raw_key, item) in enumerate(value.items()): + if index >= MAX_CONTAINER_ITEMS: + result["_truncated"] = True + break + name = _truncate_characters(self._sanitize_text(str(raw_key)), 128) + result[name] = self._sanitize_value( + item, + key=name, + depth=depth + 1, + ) + return result + if isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray)): + items = list(value[:MAX_CONTAINER_ITEMS]) + result = [ + self._sanitize_value(item, key=key, depth=depth + 1) + for item in items + ] + if len(value) > MAX_CONTAINER_ITEMS: + result.append("") + return result + return f"<{type(value).__name__}>" + + def _sanitize_text(self, value: str) -> str: + text = value.replace("\x00", "") + for target, replacement in ( + (str(self.paths.source_game), ""), + (str(self.paths.runtime_view), ""), + (str(self.paths.data), ""), + ): + if target and target != ".": + text = text.replace(target.rstrip("/"), replacement) + try: + home = str(Path(self._environment.get("HOME", "")).expanduser()) + except (OSError, RuntimeError): + home = "" + if home and home not in {".", "/"}: + text = text.replace(home.rstrip("/"), "") + text = _HOME_PATTERN.sub("", text) + text = _WINDOWS_USER_PATTERN.sub(r"\1", text) + text = _SSH_PATH_PATTERN.sub("", text) + text = _EMAIL_PATTERN.sub("", text) + text = _AUTHORIZATION_PATTERN.sub(r"\1", text) + text = _SECRET_ASSIGNMENT_PATTERN.sub(r"\1\2", text) + text = _URL_PATTERN.sub(_sanitize_url_match, text) + for private in self._private_literals: + text = re.sub( + rf"(?", + text, + ) + return text + + def _install_exception_hooks(self) -> None: + previous_sys_excepthook = sys.excepthook + self._previous_sys_excepthook = previous_sys_excepthook + + def sys_hook( + exception_type: type[BaseException], + exception: BaseException, + traceback_object: TracebackType | None, + ) -> None: + _ = exception_type, traceback_object + self.report_exception( + "launcher_exception", + exception, + context={"operation": "launcher", "stage": "uncaught_exception"}, + severity="critical", + ) + previous_sys_excepthook(exception_type, exception, traceback_object) + + sys.excepthook = sys_hook + self._installed_sys_excepthook = sys_hook + if hasattr(threading, "excepthook"): + previous_threading_excepthook = threading.excepthook + self._previous_threading_excepthook = previous_threading_excepthook + + def thread_hook(arguments: Any) -> None: + self.report_exception( + "thread_exception", + arguments.exc_value, + metadata={"thread_name": arguments.thread.name}, + context={"operation": "launcher", "stage": "worker_thread"}, + severity="critical", + ) + previous_threading_excepthook(arguments) + + threading.excepthook = thread_hook + self._installed_threading_excepthook = thread_hook + + def _restore_exception_hooks(self) -> None: + if ( + self._previous_sys_excepthook is not None + and sys.excepthook is self._installed_sys_excepthook + ): + sys.excepthook = self._previous_sys_excepthook + if ( + self._previous_threading_excepthook is not None + and threading.excepthook is self._installed_threading_excepthook + ): + threading.excepthook = self._previous_threading_excepthook + self._previous_sys_excepthook = None + self._previous_threading_excepthook = None + self._installed_sys_excepthook = None + self._installed_threading_excepthook = None + + def _debug_log(self, message: str) -> None: + if not self._debug: + return + try: + print(f"Telemetry: {message}", file=sys.stderr) + except Exception: + pass + + +_shared_service: TelemetryService | None = None +_shared_service_lock = threading.Lock() + + +def initialize_shared_telemetry( + paths: AppPaths, + *, + session_metadata: Mapping[str, object] | None = None, +) -> tuple[TelemetryService, bool]: + """Return the process-wide service and whether this caller owns its shutdown.""" + + global _shared_service + with _shared_service_lock: + if _shared_service is not None: + return _shared_service, False + service = TelemetryService( + paths, + enabled=_telemetry_enabled_from_settings(paths), + session_metadata=session_metadata, + ) + _shared_service = service + service.initialize() + return service, True + + +def current_telemetry() -> TelemetryService | None: + return _shared_service + + +def set_shared_telemetry_enabled(enabled: bool) -> None: + service = _shared_service + if service is not None: + service.set_enabled(enabled) + + +def shutdown_shared_telemetry( + service: TelemetryService, + *, + timeout_seconds: float = SHUTDOWN_WAIT_SECONDS, +) -> None: + global _shared_service + service.shutdown(timeout_seconds=timeout_seconds) + with _shared_service_lock: + if _shared_service is service: + _shared_service = None + + +def _telemetry_enabled_from_settings(paths: AppPaths) -> bool: + try: + from .settings import Settings, load_settings + + if not paths.settings_file.is_file(): + return Settings().telemetry_enabled + return load_settings(paths).telemetry_enabled + except (OSError, ValueError): + return False + + +def _post_json( + endpoint: str, + body: bytes, + headers: Mapping[str, str], + timeout_seconds: float, +) -> None: + request = Request(endpoint, data=body, headers=dict(headers), method="POST") + with urlopen(request, timeout=timeout_seconds) as response: + status = getattr(response, "status", response.getcode()) + response.read(1024) + if not isinstance(status, int) or status < 200 or status >= 300: + raise OSError(f"telemetry endpoint returned HTTP {status}") + + +def _validate_endpoint(endpoint: str) -> None: + parsed = urlsplit(endpoint) + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise ValueError("telemetry endpoint must be a credential-free HTTPS URL") + + +def _load_or_create_installation_id(path: Path) -> tuple[str, bool, str]: + path.parent.mkdir(parents=True, exist_ok=True) + try: + identity = _read_installation_id(path) + created = False + except FileNotFoundError: + candidate = str(uuid.uuid4()) + temporary = path.with_name( + f".{path.name}.{os.getpid()}.{threading.get_ident()}.{uuid.uuid4().hex}.tmp" + ) + try: + with temporary.open("x", encoding="ascii", newline="\n") as handle: + os.fchmod(handle.fileno(), _INSTALL_ID_FILE_MODE) + handle.write(candidate + "\n") + handle.flush() + os.fsync(handle.fileno()) + try: + os.link(temporary, path) + except FileExistsError: + identity = _read_installation_id(path) + created = False + else: + identity = candidate + created = True + _fsync_directory(path.parent) + finally: + temporary.unlink(missing_ok=True) + os.chmod(path, _INSTALL_ID_FILE_MODE) + first_seen = datetime.fromtimestamp(path.stat().st_mtime, timezone.utc).isoformat() + return identity, created, first_seen + + +def _read_installation_id(path: Path) -> str: + raw = path.read_text(encoding="ascii").strip() + try: + value = uuid.UUID(raw) + except (ValueError, AttributeError) as exc: + raise ValueError("invalid telemetry installation ID") from exc + if value.version != 4 or str(value) != raw.casefold(): + raise ValueError("invalid telemetry installation ID") + return str(value) + + +def _fsync_directory(path: Path) -> None: + try: + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + except OSError: + return + try: + os.fsync(descriptor) + except OSError: + pass + finally: + os.close(descriptor) + + +def _operating_system_info() -> dict[str, object]: + release = _read_os_release() + return { + "name": release.get("NAME", platform.system() or "unknown"), + "id": release.get("ID", "unknown"), + "version": release.get("PRETTY_NAME", release.get("VERSION", "unknown")), + "version_id": release.get("VERSION_ID", "unknown"), + "kernel": platform.release(), + "architecture": platform.machine() or "unknown", + "machine_architecture": platform.machine() or "unknown", + "glibc_version": platform.libc_ver()[1] or None, + } + + +def _read_os_release(path: Path = Path("/etc/os-release")) -> dict[str, str]: + try: + lines = path.read_text(encoding="utf-8", errors="replace").splitlines() + except OSError: + return {} + result: dict[str, str] = {} + for line in lines: + if not line or line.startswith("#") or "=" not in line: + continue + key, raw = line.split("=", 1) + if not re.fullmatch(r"[A-Z][A-Z0-9_]*", key): + continue + value = raw.strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: + value = value[1:-1] + value = value.replace(r"\n", " ").replace(r"\"", '"').replace(r"\\", "\\") + result[key] = value[:255] + return result + + +def _display_info(environment: Mapping[str, str]) -> dict[str, object]: + session_type = environment.get("XDG_SESSION_TYPE", "").casefold() + if session_type not in {"wayland", "x11"}: + if environment.get("WAYLAND_DISPLAY"): + session_type = "wayland" + elif environment.get("DISPLAY"): + session_type = "x11" + else: + session_type = "unknown" + desktop = environment.get("XDG_CURRENT_DESKTOP", "") + desktop = desktop.split(":", 1)[0].strip().casefold() or "unknown" + session = environment.get("DESKTOP_SESSION", "").strip().casefold() or "unknown" + return { + "server": session_type, + "desktop": desktop[:64], + "session": session[:64], + } + + +def _filesystem_info(target: Path) -> dict[str, object]: + existing = target + while not existing.exists() and existing != existing.parent: + existing = existing.parent + filesystem_type = "unknown" + readonly = False + noexec = False + try: + mounts = _read_mountinfo() + resolved = existing.resolve(strict=True) + matches = [ + mount + for mount in mounts + if resolved == mount[0] or resolved.is_relative_to(mount[0]) + ] + if matches: + mountpoint, filesystem_type, options = max( + matches, + key=lambda item: len(item[0].parts), + ) + _ = mountpoint + readonly = "ro" in options + noexec = "noexec" in options + except (OSError, RuntimeError, ValueError): + pass + try: + writable = os.access(existing, os.W_OK) and not readonly + except OSError: + writable = False + return { + "type": filesystem_type, + "writable": writable, + "readonly": readonly, + "noexec": noexec, + "network": filesystem_type.casefold() in _NETWORK_FILESYSTEMS, + } + + +def _read_mountinfo(path: Path = Path("/proc/self/mountinfo")) -> list[tuple[Path, str, set[str]]]: + mounts: list[tuple[Path, str, set[str]]] = [] + for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): + fields = line.split() + try: + separator = fields.index("-") + except ValueError: + continue + if separator < 6 or len(fields) <= separator + 3: + continue + mountpoint = Path(_unescape_mount_field(fields[4])) + filesystem_type = fields[separator + 1] + options = set(fields[5].split(",")) | set(fields[separator + 3].split(",")) + mounts.append((mountpoint, filesystem_type, options)) + return mounts + + +def _unescape_mount_field(value: str) -> str: + return re.sub( + r"\\([0-7]{3})", + lambda match: chr(int(match.group(1), 8)), + value, + ) + + +def _install_is_external(paths: AppPaths, environment: Mapping[str, str]) -> bool: + home_value = environment.get("HOME") + if not home_value: + return False + try: + home = Path(home_value).expanduser().resolve() + data = paths.data.resolve() + return data != home and not data.is_relative_to(home) + except (OSError, RuntimeError): + return False + + +def _current_game_version(path: Path) -> str | None: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(value, dict): + return None + version = value.get("version") + return version[:32] if isinstance(version, str) and version else None + + +def _settings_schema_version() -> int: + try: + from .settings import SETTINGS_SCHEMA + + return SETTINGS_SCHEMA + except Exception: + return 1 + + +def _is_secret_key(key: str) -> bool: + normalized = key.casefold().replace("-", "_").strip() + return normalized in _SECRET_KEYS + + +def _string_limit_for_key(key: str | None) -> int: + normalized = (key or "").casefold() + if normalized == "stderr": + return MAX_STDERR_CHARACTERS + if normalized == "stdout": + return MAX_STDOUT_CHARACTERS + if normalized in {"sanitized_traceback", "traceback"}: + return MAX_TRACEBACK_CHARACTERS + return 8192 + + +def _keep_string_tail(key: str | None) -> bool: + return (key or "").casefold() in {"stderr", "stdout", "sanitized_traceback", "traceback"} + + +def _sanitize_url_match(match: re.Match[str]) -> str: + raw = match.group(0) + trailing = "" + while raw and raw[-1] in ".,;:)]": + trailing = raw[-1] + trailing + raw = raw[:-1] + try: + parsed = urlsplit(raw) + except ValueError: + return "" + trailing + host = parsed.hostname or "" + if not host: + return "" + trailing + netloc = host + if ":" in host and not host.startswith("["): + netloc = f"[{host}]" + try: + if parsed.port is not None: + netloc += f":{parsed.port}" + except ValueError: + return "" + trailing + query = parsed.query + if parsed.username is not None or any( + str(key).casefold() in _SECRET_QUERY_KEYS for key, _value in parse_qsl(query) + ): + query = "" if query else "" + return urlunsplit((parsed.scheme, netloc, parsed.path, query, "")) + trailing + + +def _truncate_characters(value: str, limit: int, *, tail: bool = False) -> str: + if len(value) <= limit: + return value + marker = "" + remaining = max(0, limit - len(marker)) + return marker + value[-remaining:] if tail else value[:remaining] + marker + + +def _truncate_utf8(value: str, limit: int, *, tail: bool = False) -> str: + encoded = value.encode("utf-8") + if len(encoded) <= limit: + return value + marker = b"" + remaining = max(0, limit - len(marker)) + chunk = encoded[-remaining:] if tail else encoded[:remaining] + decoded = chunk.decode("utf-8", errors="ignore") + return "" + decoded if tail else decoded + "" + + +def _bound_mapping(value: dict[str, object], limit: int) -> dict[str, object]: + if len(_json_bytes(value)) <= limit: + return value + bounded: dict[str, object] = {} + for key, item in value.items(): + candidate = {**bounded, key: item, "_truncated": True} + if len(_json_bytes(candidate)) <= limit: + bounded[key] = item + continue + if isinstance(item, str): + for string_limit in (4096, 2048, 1024, 512, 256, 128): + shortened = _truncate_utf8(item, string_limit, tail=_keep_string_tail(key)) + candidate = {**bounded, key: shortened, "_truncated": True} + if len(_json_bytes(candidate)) <= limit: + bounded[key] = shortened + break + bounded["_truncated"] = True + while len(_json_bytes(bounded)) > limit and len(bounded) > 1: + last = next(key for key in reversed(bounded) if key != "_truncated") + bounded.pop(last) + return bounded + + +def _encode_payload(payload: dict[str, object]) -> bytes | None: + body = _json_bytes(payload) + if len(body) <= MAX_REQUEST_BYTES: + return body + for key in ("metadata", "context"): + if key in payload: + payload[key] = {"_truncated": True} + body = _json_bytes(payload) + if len(body) <= MAX_REQUEST_BYTES: + return body + if isinstance(payload.get("message"), str): + payload["message"] = _truncate_utf8(str(payload["message"]), 2048) + body = _json_bytes(payload) + return body if len(body) <= MAX_REQUEST_BYTES else None + + +def _json_bytes(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + + +def normalized_error_message(message: str) -> str: + """Normalize dynamic values for stable server-compatible fingerprints.""" + + value = _UUID_PATTERN.sub("", message) + value = _MEMORY_ADDRESS_PATTERN.sub("
", value) + return re.sub(r"\b\d{4}-\d{2}-\d{2}[T ][0-9:.+-Z]+\b", "", value) + + +def _resource_usage() -> dict[str, object]: + """Read one bounded process-usage snapshot on the Linux launcher host.""" + + try: + usage = resource.getrusage(resource.RUSAGE_SELF) + peak_value = float(usage.ru_maxrss) + peak_divisor = 1024 * 1024 if sys.platform == "darwin" else 1024 + peak_memory_mb = peak_value / peak_divisor + cpu_time_seconds = float(usage.ru_utime) + float(usage.ru_stime) + except (OSError, TypeError, ValueError, OverflowError): + return {} + if ( + not math.isfinite(peak_memory_mb) + or not math.isfinite(cpu_time_seconds) + or peak_memory_mb < 0 + or cpu_time_seconds < 0 + ): + return {} + return { + "peak_memory_mb": round(peak_memory_mb, 2), + "cpu_time_seconds": round(cpu_time_seconds, 3), + } + + +def _is_error_event(event: str, severity: str) -> bool: + return ( + severity in {"error", "critical"} + or "failed" in event + or "exception" in event + or "error" in event + ) + + +def _string_field(values: Mapping[str, object], key: str) -> str: + value = values.get(key) + return value if isinstance(value, str) else "" + + +def error_fingerprint( + event: str, + message: str, + *, + exception_type: str = "", + source_module: str = "", + source_function: str = "", +) -> str: + """Create a privacy-safe deterministic fingerprint for tests and server parity.""" + + material = "\n".join( + (event, exception_type, source_module, source_function, normalized_error_message(message)) + ) + return hashlib.sha256(material.encode("utf-8", errors="replace")).hexdigest() + + +def classify_network_failure(error: object) -> str: + """Return one stable low-cardinality class for download failures.""" + + message = str(error).casefold() + name = type(error).__name__.casefold() + if "timeout" in name or "timed out" in message or "timeout" in message: + return "connection_timeout" + if any( + value in message + for value in ( + "name or service not known", + "name resolution", + "nodename nor servname", + "temporary failure in name", + ) + ): + return "dns_failure" + if "connection refused" in message: + return "connection_refused" + if "connection reset" in message or "broken pipe" in message: + return "connection_reset" + if any(value in message for value in ("certificate", "ssl", "tls")): + return "tls_failure" + if re.search(r"\bhttp\s+\d{3}\b", message): + return "http_error" + if any( + value in message + for value in ("sha-1 mismatch", "sha-256", "checksum", "content-integrity") + ): + return "content_integrity" + if "could not connect" in message or "unreachable" in message: + return "cdn_unreachable" + return "unknown" + + +def network_http_status(error: object) -> int | None: + """Extract an HTTP status without exposing a failed request URL.""" + + current: object | None = error + seen: set[int] = set() + while current is not None and id(current) not in seen: + seen.add(id(current)) + status = getattr(current, "code", None) + if isinstance(status, int) and 100 <= status <= 599: + return status + match = _HTTP_STATUS_PATTERN.search(str(current)) + if match is not None: + return int(match.group(1)) + current = getattr(current, "__cause__", None) or getattr( + current, + "__context__", + None, + ) + return None diff --git a/ptlinux/telemetry_environment.py b/ptlinux/telemetry_environment.py new file mode 100644 index 0000000..c4b3076 --- /dev/null +++ b/ptlinux/telemetry_environment.py @@ -0,0 +1,1221 @@ +"""Bounded, privacy-safe compatibility details for launcher telemetry.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +import json +import os +from pathlib import Path +import platform +import re +import shutil +import struct +import subprocess + +from .paths import AppPaths + + +PROBE_TIMEOUT_SECONDS = 3.0 +MAX_MARKER_BYTES = 256 * 1024 +MAX_DLL_SCAN_BYTES = 16 * 1024 * 1024 +MAX_PROBE_OUTPUT_CHARACTERS = 64 * 1024 + +_VULKAN_MANIFEST_DIRECTORIES = ( + Path("/etc/vulkan/icd.d"), + Path("/usr/local/share/vulkan/icd.d"), + Path("/usr/share/vulkan/icd.d"), +) +_LIBRARY_ROOTS = ( + Path("/lib"), + Path("/usr/lib"), + Path("/usr/local/lib"), +) +_PCI_DATABASES = ( + Path("/usr/share/hwdata/pci.ids"), + Path("/usr/share/misc/pci.ids"), + Path("/usr/share/pci.ids"), +) +_GPU_VENDORS = { + 0x1002: "AMD", + 0x10DE: "NVIDIA", + 0x8086: "Intel", +} +_DXVK_VERSION_PATTERN = re.compile( + r"\bdxvk(?:\s+version|[- /:])?\s*v?(\d+\.\d+(?:\.\d+){0,2})\b", + re.IGNORECASE, +) +_VERSION_PATTERN = re.compile(r"\d+(?:\.\d+){1,3}") + + +@dataclass(frozen=True) +class ProbeResult: + """Small result surface for bounded local compatibility probes.""" + + returncode: int + stdout: str + stderr: str + + +ProbeRunner = Callable[[Sequence[str], Mapping[str, str], float], ProbeResult | None] + + +def collect_extended_environment( + paths: AppPaths, + environment: Mapping[str, str], + *, + probe_runner: ProbeRunner | None = None, + sysfs_drm: Path = Path("/sys/class/drm"), + vulkan_manifest_directories: Sequence[Path] = _VULKAN_MANIFEST_DIRECTORIES, + library_roots: Sequence[Path] = _LIBRARY_ROOTS, + pci_databases: Sequence[Path] = _PCI_DATABASES, + now: datetime | None = None, +) -> dict[str, object]: + """Collect compatibility data once without returning paths or device IDs.""" + + run_probe = probe_runner or _run_probe + marker = _read_json_object(paths.game_runtime_marker) + prefix = _selected_prefix(paths, marker) + runtime = _runtime_info( + paths, + marker, + prefix, + environment, + run_probe, + vulkan_manifest_directories=vulkan_manifest_directories, + library_roots=library_roots, + now=now or datetime.now(timezone.utc), + ) + gpu = _gpu_info( + runtime.pop("_vulkan_summary", None), + environment=environment, + sysfs_drm=sysfs_drm, + pci_databases=pci_databases, + ) + dependencies = _dependency_health( + environment, + prefix, + runtime, + ) + result: dict[str, object] = { + "runtime": runtime, + "dependencies": dependencies, + } + display = _display_info(sysfs_drm) + if display: + result["display"] = display + if gpu: + result["gpu"] = gpu + return result + + +def _runtime_info( + paths: AppPaths, + marker: Mapping[str, object] | None, + prefix: Path | None, + environment: Mapping[str, str], + probe_runner: ProbeRunner, + *, + vulkan_manifest_directories: Sequence[Path], + library_roots: Sequence[Path], + now: datetime, +) -> dict[str, object]: + runtime_marker = _object_field(marker, "runtime") + runtime_kind = _short_string(runtime_marker.get("kind")) if runtime_marker else None + result: dict[str, object] = { + "kind": runtime_kind or "unknown", + "wine_vendor": _wine_vendor(runtime_kind, runtime_marker, paths), + } + wine_command = _wine_command(runtime_marker, environment) + if wine_command is not None: + wine_probe = probe_runner( + (wine_command, "--version"), + environment, + PROBE_TIMEOUT_SECONDS, + ) + wine_build = _first_probe_line(wine_probe) + if wine_build: + result["wine_build"] = wine_build + wine_version = _wine_version(wine_build) + if wine_version: + result["wine"] = wine_version + + if runtime_marker: + proton_name = _proton_name(runtime_marker.get("proton_path")) + if proton_name: + result["proton_version"] = proton_name + + if prefix is not None: + prefix_architecture = _wine_prefix_architecture(prefix) + if prefix_architecture: + result["prefix_arch"] = prefix_architecture + configured_at = _marker_timestamp(marker) + if configured_at is not None: + result["prefix_created_at"] = configured_at.isoformat() + result["prefix_age_days"] = max(0, (now - configured_at).days) + result.update( + _dxvk_info( + prefix, + packaged_version=_proton_dxvk_version(runtime_marker), + ) + ) + else: + result.update( + { + "dxvk_enabled": False, + "dxvk_install_status": "missing", + } + ) + result.update(_runtime_component_info(paths.launcher_prefix, prefix)) + + vulkan = _vulkan_info( + environment, + probe_runner, + manifest_directories=vulkan_manifest_directories, + library_roots=library_roots, + ) + result.update(vulkan) + return result + + +def _vulkan_info( + environment: Mapping[str, str], + probe_runner: ProbeRunner, + *, + manifest_directories: Sequence[Path], + library_roots: Sequence[Path], +) -> dict[str, object]: + loader_architectures = _vulkan_loader_architectures(library_roots) + manifest_architectures: set[int] = set() + api_versions: list[str] = [] + for manifest in _vulkan_manifests(environment, manifest_directories): + parsed = _read_json_object(manifest) + icd = _object_field(parsed, "ICD") + if icd is None: + continue + api_version = _short_string(icd.get("api_version")) + if api_version and _VERSION_PATTERN.fullmatch(api_version): + api_versions.append(api_version) + library = _short_string(icd.get("library_path")) + architecture = _manifest_library_architecture( + manifest, + library, + library_roots, + ) + if architecture is not None: + manifest_architectures.add(architecture) + + vulkaninfo = shutil.which("vulkaninfo", path=environment.get("PATH", "")) + probe = ( + probe_runner( + (vulkaninfo, "--summary"), + environment, + PROBE_TIMEOUT_SECONDS, + ) + if vulkaninfo + else None + ) + functional_64bit = probe is not None and probe.returncode == 0 + summary = _probe_text(probe) if functional_64bit else "" + if summary: + api_version = _summary_value(summary, "apiVersion") + if api_version: + api_versions.append(api_version) + instance_version = re.search( + r"Vulkan\s+Instance\s+Version\s*:\s*(\S+)", + summary, + re.IGNORECASE, + ) + if instance_version: + api_versions.append(instance_version.group(1)) + + available_64bit = functional_64bit or ( + 64 in loader_architectures and 64 in manifest_architectures + ) + available_32bit = ( + 32 in loader_architectures and 32 in manifest_architectures + ) + result: dict[str, object] = { + "vulkan_available": available_64bit, + "vulkan_64bit": available_64bit, + "vulkan_32bit": available_32bit, + "vulkan_64bit_detection": ( + "vulkaninfo" if functional_64bit else "loader_and_icd" + ), + "vulkan_32bit_detection": "loader_and_icd", + } + best_api_version = _maximum_version(api_versions) + if best_api_version: + result["vulkan_api_version"] = best_api_version + driver_info = _summary_value(summary, "driverInfo") + if driver_info: + result["vulkan_driver_version"] = _driver_version(driver_info) or driver_info + if summary: + result["_vulkan_summary"] = summary + return result + + +def _gpu_info( + vulkan_summary: object, + *, + environment: Mapping[str, str], + sysfs_drm: Path, + pci_databases: Sequence[Path], +) -> dict[str, object]: + summary = vulkan_summary if isinstance(vulkan_summary, str) else "" + vulkan_gpus = _vulkan_gpu_summaries(summary) + has_gpu_blocks = re.search(r"(?mi)^\s*GPU\d+\s*:\s*$", summary) is not None + if vulkan_gpus: + primary_summary = vulkan_gpus[0] + elif has_gpu_blocks: + primary_summary = "" + else: + primary_summary = summary + vendor_id = _hex_summary_value(primary_summary, "vendorID") + model = _summary_value(primary_summary, "deviceName") + driver_name = _summary_value(primary_summary, "driverName") + driver_info = _summary_value(primary_summary, "driverInfo") + sysfs_gpus = _sysfs_gpus(sysfs_drm) + sysfs = next( + (gpu for gpu in sysfs_gpus if gpu.get("boot_vga") is True), + sysfs_gpus[0] if sysfs_gpus else None, + ) + if vendor_id is None and sysfs is not None: + vendor_id = sysfs.get("vendor_id") + if model is None and sysfs is not None: + model = _short_string(sysfs.get("model")) + if model is None and sysfs is not None: + device_id = sysfs.get("device_id") + if isinstance(vendor_id, int) and isinstance(device_id, int): + model = _pci_model(vendor_id, device_id, pci_databases) + if driver_name is None and sysfs is not None: + driver_name = _short_string(sysfs.get("driver")) + if driver_info is None and sysfs is not None: + driver_info = _short_string(sysfs.get("driver_version")) + + vendor = _GPU_VENDORS.get(vendor_id) if isinstance(vendor_id, int) else None + if vendor is None: + vendor = _vendor_from_model(model) + driver = _friendly_driver_name(driver_name, driver_info) + driver_version = _driver_version(driver_info) + result: dict[str, object] = {} + if vendor: + result["vendor"] = vendor + if model: + result["model"] = model[:255] + if driver: + result["driver"] = driver[:128] + if driver_version: + result["driver_version"] = driver_version[:128] + if sysfs is not None and isinstance(sysfs.get("vram_mb"), int): + result["vram_mb"] = sysfs["vram_mb"] + summary_gpu_count = len(vulkan_gpus) + if summary_gpu_count == 0 and model: + summary_gpu_count = 1 + gpu_count = max(len(sysfs_gpus), summary_gpu_count) + if gpu_count: + result["count"] = gpu_count + if vendor: + result["primary_vendor"] = vendor + if model: + result["primary_model"] = model[:255] + vendor_ids = { + vendor + for gpu in sysfs_gpus + if isinstance((vendor := gpu.get("vendor_id")), int) + } + for gpu_summary in vulkan_gpus: + vendor_ids.update(_hex_summary_values(gpu_summary, "vendorID")) + device_types: set[str] = set() + for gpu_summary in vulkan_gpus: + device_type = _summary_value(gpu_summary, "deviceType") + if isinstance(device_type, str): + device_types.add(device_type.casefold()) + integrated_and_discrete = ( + any("integrated" in value for value in device_types) + and any("discrete" in value for value in device_types) + ) + result["hybrid_graphics"] = gpu_count > 1 and ( + len(vendor_ids) > 1 or integrated_and_discrete + ) + result["offload_detected"] = _gpu_offload_detected(environment) + return result + + +def _dependency_health( + environment: Mapping[str, str], + prefix: Path | None, + runtime: Mapping[str, object], +) -> dict[str, object]: + tools: dict[str, bool] = {} + required: list[str] = [] + missing: list[str] = [] + container_ready: bool | None = None + try: + from .doctor import inspect_host + + report = inspect_host(environment) + for item in report.tools: + available = item.path is not None + tools[item.name] = available + if item.required: + required.append(item.name) + if not available: + missing.append(item.name) + container_ready = report.container_host.ready + report_ok: bool | None = report.ok + except Exception: + report_ok = None + + wine_available = tools.get("wine") + if wine_available is None: + wine_available = _wine_command(None, environment) is not None + wine_32bit = _wine_32bit_available(prefix) + vulkan_64bit = _optional_bool(runtime.get("vulkan_64bit")) + vulkan_32bit = _optional_bool(runtime.get("vulkan_32bit")) + font_dependencies = _font_dependencies_present(prefix) + archive_tools = tools.get("cabextract") + if archive_tools is None: + archive_tools = shutil.which( + "cabextract", + path=environment.get("PATH", ""), + ) is not None + shared_libraries = ( + vulkan_64bit and vulkan_32bit + if platform.machine().casefold() in {"x86_64", "amd64"} + else vulkan_64bit + ) + healthy = report_ok + if healthy is not None: + healthy = bool(healthy and shared_libraries) + result: dict[str, object] = { + "healthy": healthy, + "required_count": len(required), + "available_required_count": len(required) - len(missing), + "missing_required": missing, + "tools": tools, + "wine_available": wine_available, + "wine_32bit_available": wine_32bit, + "vulkan_available": bool(runtime.get("vulkan_available")), + "vulkan_64bit_available": vulkan_64bit, + "vulkan_32bit_available": vulkan_32bit, + "font_dependencies_present": font_dependencies, + "dotnet_present": bool(runtime.get("dotnet_present")), + "archive_tools_present": archive_tools, + "required_shared_libraries_present": shared_libraries, + "directx_runtime_present": bool(runtime.get("directx_runtime_present")), + } + if container_ready is not None: + result["container_ready"] = container_ready + return result + + +def _selected_prefix( + paths: AppPaths, + marker: Mapping[str, object] | None, +) -> Path | None: + runtime = _object_field(marker, "runtime") + kind = _short_string(runtime.get("kind")) if runtime else None + preferred = paths.umu_prefix if kind == "umu" else paths.proton_prefix + for candidate in (preferred, paths.proton_prefix, paths.umu_prefix): + if (candidate / "system.reg").is_file() or ( + candidate / "drive_c/windows" + ).is_dir(): + return candidate + return None + + +def _wine_command( + runtime: Mapping[str, object] | None, + environment: Mapping[str, str], +) -> str | None: + if runtime: + proton_path = _short_string(runtime.get("proton_path")) + if proton_path: + root = Path(proton_path).expanduser().parent + for relative in ( + "files/bin/wine64", + "files/bin/wine", + "dist/bin/wine64", + "dist/bin/wine", + ): + candidate = root / relative + if candidate.is_file() and os.access(candidate, os.X_OK): + return str(candidate) + executable = _short_string(runtime.get("executable")) + if executable and Path(executable).name.casefold() in {"wine", "wine64"}: + return executable + return shutil.which("wine", path=environment.get("PATH", "")) + + +def _wine_vendor( + runtime_kind: str | None, + runtime: Mapping[str, object] | None, + paths: AppPaths, +) -> str: + if runtime_kind in {"proton7", "proton7-container", "umu"}: + return "proton" if runtime_kind != "umu" else "umu" + if runtime: + executable = _short_string(runtime.get("executable")) + if executable: + lowered = executable.casefold() + if "lutris" in lowered: + return "lutris" + try: + if Path(executable).resolve().is_relative_to(paths.data.resolve()): + return "pristontale-runtime" + except (OSError, RuntimeError): + pass + return "system" + + +def _wine_version(build: str) -> str | None: + match = re.search(r"\bwine(?:64)?[- ]v?(\d+(?:\.\d+){1,3})", build, re.IGNORECASE) + if match: + return match.group(1) + match = _VERSION_PATTERN.search(build) + return match.group(0) if match else None + + +def _proton_name(value: object) -> str | None: + path = _short_string(value) + if not path: + return None + name = Path(path).parent.name.strip() + if "proton" not in name.casefold(): + return None + version = re.search(r"\d+(?:[._-]\d+)+(?:[A-Za-z0-9._-]*)?", name) + return version.group(0)[:64] if version else None + + +def _wine_prefix_architecture(prefix: Path) -> str | None: + try: + with (prefix / "system.reg").open("r", encoding="utf-8", errors="replace") as handle: + text = handle.read(4096) + except OSError: + return None + match = re.search(r"(?m)^#arch=(win32|win64)\s*$", text, re.IGNORECASE) + return match.group(1).casefold() if match else None + + +def _marker_timestamp(marker: Mapping[str, object] | None) -> datetime | None: + if marker is None: + return None + value = _short_string(marker.get("configured_at")) + if not value: + return None + try: + parsed = datetime.fromisoformat(value) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _dxvk_info( + prefix: Path, + *, + packaged_version: str | None = None, +) -> dict[str, object]: + components: dict[str, str | None] = {} + identified_components: set[str] = set() + files_present = 0 + dxvk_components = 0 + invalid = 0 + for name in ("d3d9", "d3d11", "dxgi"): + versions: list[str] = [] + component_present = False + component_is_dxvk = False + for system_directory in ("syswow64", "system32"): + dll = prefix / "drive_c/windows" / system_directory / f"{name}.dll" + if dll.is_symlink() and not dll.exists(): + invalid += 1 + continue + if not dll.is_file(): + continue + component_present = True + is_dxvk, version = _dxvk_dll_details(dll) + component_is_dxvk = component_is_dxvk or is_dxvk + if version: + versions.append(version) + if component_present: + files_present += 1 + if component_is_dxvk: + dxvk_components += 1 + identified_components.add(name) + components[name] = _maximum_version(versions) + + if invalid: + status = "invalid" + elif files_present == 0: + status = "missing" + elif dxvk_components == 0: + status = "unknown" + elif dxvk_components < len(components): + status = "partial" + else: + status = "installed" + result: dict[str, object] = { + "dxvk_enabled": dxvk_components > 0 and status != "invalid", + "dxvk_install_status": status, + } + all_versions = [ + version for version in components.values() if isinstance(version, str) + ] + dxvk_version = _maximum_version(all_versions) or packaged_version + if dxvk_version: + result["dxvk_version"] = dxvk_version + for name, version in components.items(): + if version is None and name in identified_components: + version = packaged_version + if version: + result[f"dxvk_{name}_version"] = version + return result + + +def _proton_dxvk_version(runtime: Mapping[str, object] | None) -> str | None: + if runtime is None: + return None + proton_path = _short_string(runtime.get("proton_path")) + if not proton_path: + return None + root = Path(proton_path).expanduser().parent + for relative in ( + "dist/lib/wine/dxvk/version", + "dist/lib64/wine/dxvk/version", + "files/lib/wine/dxvk/version", + "files/lib64/wine/dxvk/version", + ): + value = _read_short_text(root / relative) + if not value: + continue + match = re.search( + r"\bdxvk\s+\(v([0-9][0-9A-Za-z.+-]{1,63})\)", + value, + re.IGNORECASE, + ) + if match: + return match.group(1) + return None + + +def _dxvk_dll_details(path: Path) -> tuple[bool, str | None]: + try: + with path.open("rb") as handle: + content = handle.read(MAX_DLL_SCAN_BYTES + 1) + except OSError: + return False, None + if len(content) > MAX_DLL_SCAN_BYTES: + content = content[:MAX_DLL_SCAN_BYTES] + texts = ( + content.decode("latin-1", errors="ignore"), + content.decode("utf-16-le", errors="ignore"), + ) + versions: list[str] = [] + is_dxvk = False + for text in texts: + is_dxvk = is_dxvk or "dxvk" in text.casefold() + versions.extend(match.group(1) for match in _DXVK_VERSION_PATTERN.finditer(text)) + version = _maximum_version(versions) + if version is None and is_dxvk: + version = _pe_fixed_file_version(content) + return is_dxvk, version + + +def _pe_fixed_file_version( + content: bytes, + *, + maximum_part: int = 10_000, +) -> str | None: + """Read the numeric VS_FIXEDFILEINFO version without a PE dependency.""" + + signature = b"\xbd\x04\xef\xfe" + offset = content.find(signature) + while offset >= 0 and offset + 52 <= len(content): + try: + values = struct.unpack_from("<13I", content, offset) + except struct.error: + return None + structure_version = values[1] + major = values[2] >> 16 + minor = values[2] & 0xFFFF + build = values[3] >> 16 + revision = values[3] & 0xFFFF + parts = [major, minor, build, revision] + if structure_version >> 16 == 1 and 0 < major < 100 and all( + part < maximum_part for part in parts + ): + while len(parts) > 2 and parts[-1] == 0: + parts.pop() + return ".".join(str(part) for part in parts) + offset = content.find(signature, offset + len(signature)) + return None + + +def _directx_runtime_present(prefix: Path) -> bool: + directory = prefix / "drive_c/windows/syswow64" + return all((directory / f"{name}.dll").is_file() for name in ("d3dx9_43", "d3dcompiler_43")) + + +def _runtime_component_info( + launcher_prefix: Path, + game_prefix: Path | None, +) -> dict[str, object]: + prefixes = tuple( + dict.fromkeys( + prefix + for prefix in (game_prefix, launcher_prefix) + if prefix is not None + ) + ) + dotnet_runtime_versions: list[str] = [] + dotnet_desktop_versions: list[str] = [] + dotnet_framework_versions: list[str] = [] + for prefix in prefixes: + dotnet_root = prefix / "drive_c/Program Files/dotnet/shared" + dotnet_runtime_versions.extend( + _directory_versions(dotnet_root / "Microsoft.NETCore.App") + ) + dotnet_desktop_versions.extend( + _directory_versions(dotnet_root / "Microsoft.WindowsDesktop.App") + ) + for framework_name in ("Framework", "Framework64"): + dotnet_framework_versions.extend( + _directory_versions( + prefix / "drive_c/windows/Microsoft.NET" / framework_name + ) + ) + + dotnet_runtime = _maximum_version(dotnet_runtime_versions) + dotnet_desktop = _maximum_version(dotnet_desktop_versions) + dotnet_framework = _maximum_version(dotnet_framework_versions) + vc_present, vc_version = _vc_runtime_info(prefixes) + directx_present = ( + _directx_runtime_present(game_prefix) + if game_prefix is not None + else False + ) + winetricks_components: list[str] = [] + if game_prefix is not None: + system = game_prefix / "drive_c/windows/syswow64" + for name in ("d3dcompiler_43", "d3dx9_43"): + if (system / f"{name}.dll").is_file(): + winetricks_components.append(name) + if dotnet_desktop: + winetricks_components.append( + "dotnetdesktop6" + if dotnet_desktop.startswith("6.") + else "dotnetdesktop" + ) + if vc_present: + winetricks_components.append("vcrun2015-2022") + + runtime_components: dict[str, object] = { + "directx": directx_present, + "vc_runtime": vc_present, + } + if dotnet_runtime or dotnet_framework: + runtime_components["dotnet_runtime"] = dotnet_runtime or dotnet_framework + if dotnet_desktop: + runtime_components["dotnet_desktop"] = dotnet_desktop + if vc_version: + runtime_components["vc_runtime_version"] = vc_version + + result: dict[str, object] = { + "dotnet_present": bool( + dotnet_runtime + or dotnet_desktop + or dotnet_framework + or (game_prefix is not None and _dotnet_present(game_prefix)) + ), + "vc_runtime_present": vc_present, + "directx_runtime_present": directx_present, + "winetricks_components": winetricks_components, + "runtime_components": runtime_components, + } + if dotnet_runtime or dotnet_framework: + result["dotnet_runtime_version"] = dotnet_runtime or dotnet_framework + if dotnet_desktop: + result["dotnet_desktop_runtime_version"] = dotnet_desktop + if vc_version: + result["vc_runtime_version"] = vc_version + return result + + +def _directory_versions(root: Path) -> tuple[str, ...]: + try: + iterator = root.iterdir() + except OSError: + return () + versions: list[str] = [] + try: + for index, entry in enumerate(iterator): + if index >= 64: + break + try: + is_directory = entry.is_dir() + except OSError: + continue + if not is_directory: + continue + version = entry.name.removeprefix("v") + if re.fullmatch(r"\d+(?:\.\d+){0,3}", version): + versions.append(version) + except OSError: + return tuple(versions) + return tuple(versions) + + +def _vc_runtime_info(prefixes: Sequence[Path]) -> tuple[bool, str | None]: + versions: list[str] = [] + present = False + for prefix in prefixes: + for system_directory in ("syswow64", "system32"): + system = prefix / "drive_c/windows" / system_directory + for name in ("vcruntime140.dll", "vcruntime140_1.dll", "msvcp140.dll"): + dll = system / name + if not dll.is_file(): + continue + present = True + version = _dll_file_version(dll) + if version: + versions.append(version) + return present, _maximum_version(versions) + + +def _dll_file_version(path: Path) -> str | None: + try: + with path.open("rb") as handle: + content = handle.read(MAX_DLL_SCAN_BYTES) + except OSError: + return None + return _pe_fixed_file_version(content, maximum_part=1_000_000) + + +def _dotnet_present(prefix: Path) -> bool: + framework = prefix / "drive_c/windows/Microsoft.NET" + try: + return any(path.is_file() for path in framework.glob("Framework*/v*/mscorlib.dll")) + except OSError: + return False + + +def _wine_32bit_available(prefix: Path | None) -> bool | None: + if prefix is None: + return None + architecture = _wine_prefix_architecture(prefix) + if architecture == "win32": + return True + return (prefix / "drive_c/windows/syswow64").is_dir() + + +def _font_dependencies_present(prefix: Path | None) -> bool: + candidates = [ + Path("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"), + Path("/usr/share/fonts/truetype/liberation2/LiberationSans-Regular.ttf"), + ] + if prefix is not None: + fonts = prefix / "drive_c/windows/Fonts" + candidates.extend((fonts / "arial.ttf", fonts / "tahoma.ttf")) + return any(candidate.is_file() for candidate in candidates) + + +def _vulkan_manifests( + environment: Mapping[str, str], + directories: Sequence[Path], +) -> tuple[Path, ...]: + candidates: list[Path] = [] + explicit = environment.get("VK_ICD_FILENAMES", "") + for value in explicit.split(os.pathsep): + if value: + candidates.append(Path(value).expanduser()) + for directory in directories: + try: + candidates.extend(sorted(directory.glob("*.json"))) + except OSError: + continue + seen: set[Path] = set() + result: list[Path] = [] + for candidate in candidates: + try: + resolved = candidate.resolve() + except (OSError, RuntimeError): + continue + if resolved in seen or not resolved.is_file(): + continue + seen.add(resolved) + result.append(resolved) + return tuple(result) + + +def _vulkan_loader_architectures(library_roots: Sequence[Path]) -> set[int]: + architectures: set[int] = set() + for root in library_roots: + for candidate in _library_candidates(root, "libvulkan.so"): + architecture = _elf_architecture(candidate) + if architecture is not None: + architectures.add(architecture) + return architectures + + +def _manifest_library_architecture( + manifest: Path, + library: str | None, + library_roots: Sequence[Path], +) -> int | None: + if library: + path = Path(library) + if path.is_absolute(): + architecture = _elf_architecture(path) + if architecture is not None: + return architecture + else: + candidates = [manifest.parent / path] + for root in library_roots: + candidates.extend(_library_candidates(root, path.name)) + for candidate in candidates: + architecture = _elf_architecture(candidate) + if architecture is not None: + return architecture + lowered = manifest.name.casefold() + if any(token in lowered for token in ("i386", "i486", "i586", "i686", "x86_32")): + return 32 + if any(token in lowered for token in ("x86_64", "amd64")): + return 64 + return None + + +def _library_candidates(root: Path, prefix: str) -> tuple[Path, ...]: + try: + direct = tuple(root.glob(f"{prefix}*")) + children = tuple(root.glob(f"*/{prefix}*")) + except OSError: + return () + return direct + children + + +def _elf_architecture(path: Path) -> int | None: + try: + with path.open("rb") as handle: + header = handle.read(5) + except OSError: + return None + if len(header) < 5 or header[:4] != b"\x7fELF": + return None + if header[4] == 1: + return 32 + if header[4] == 2: + return 64 + return None + + +def _sysfs_gpus(sysfs_drm: Path) -> tuple[dict[str, object], ...]: + try: + cards = sorted( + path for path in sysfs_drm.glob("card[0-9]*") if path.name[4:].isdigit() + ) + except OSError: + return () + result_gpus: list[dict[str, object]] = [] + for card in cards[:16]: + device = card / "device" + vendor_id = _read_hex(device / "vendor") + device_id = _read_hex(device / "device") + if vendor_id is None or device_id is None: + continue + result: dict[str, object] = { + "vendor_id": vendor_id, + "device_id": device_id, + } + if _read_decimal(device / "boot_vga") == 1: + result["boot_vga"] = True + for name in ("product_name", "label"): + value = _read_short_text(device / name) + if value: + result["model"] = value + break + try: + driver = (device / "driver").resolve(strict=True).name + except (OSError, RuntimeError): + driver = None + if driver: + result["driver"] = driver + version = _read_short_text(Path("/sys/module") / driver / "version") + if version: + result["driver_version"] = version + vram_bytes = _read_decimal(device / "mem_info_vram_total") + if vram_bytes is not None and vram_bytes > 0: + result["vram_mb"] = round(vram_bytes / (1024 * 1024)) + result_gpus.append(result) + return tuple(result_gpus) + + +def _display_info(sysfs_drm: Path) -> dict[str, object]: + monitor_count = _connected_monitor_count(sysfs_drm) + return {"monitor_count": monitor_count} if monitor_count is not None else {} + + +def _connected_monitor_count(sysfs_drm: Path) -> int | None: + try: + connectors = sorted(sysfs_drm.glob("card[0-9]*-*/status"))[:32] + except OSError: + return None + if not connectors: + return None + connected = 0 + for status in connectors: + value = _read_short_text(status) + if value and value.casefold() == "connected": + connected += 1 + return connected + + +def _pci_model( + vendor_id: int, + device_id: int, + databases: Sequence[Path], +) -> str | None: + vendor_key = f"{vendor_id:04x}" + device_key = f"{device_id:04x}" + for database in databases: + try: + lines = database.read_text(encoding="utf-8", errors="replace").splitlines() + except OSError: + continue + inside_vendor = False + for line in lines: + if not line or line.startswith("#"): + continue + if not line.startswith("\t"): + parts = line.split(None, 1) + inside_vendor = bool(parts and parts[0].casefold() == vendor_key) + continue + if not inside_vendor or line.startswith("\t\t"): + continue + parts = line.strip().split(None, 1) + if len(parts) == 2 and parts[0].casefold() == device_key: + return parts[1].strip()[:255] + return None + + +def _friendly_driver_name(name: str | None, info: str | None) -> str | None: + lowered = (name or "").casefold() + combined = f"{name or ''} {info or ''}".casefold() + if "radv" in combined: + return "Mesa RADV" + if "anv" in combined or "intel open-source mesa" in combined: + return "Mesa ANV" + if "nvidia" in combined: + return "NVIDIA" + if "nouveau" in combined: + return "Mesa NVK" if "nvk" in combined else "Nouveau" + if lowered in {"amdgpu", "i915", "xe"}: + return lowered + return name.strip() if name and name.strip() else None + + +def _driver_version(info: str | None) -> str | None: + if not info: + return None + mesa = re.search(r"\bMesa\s+(\d+(?:\.\d+){1,3})", info, re.IGNORECASE) + if mesa: + return mesa.group(1) + match = _VERSION_PATTERN.search(info) + return match.group(0) if match else None + + +def _vendor_from_model(model: str | None) -> str | None: + lowered = (model or "").casefold() + if "amd" in lowered or "radeon" in lowered: + return "AMD" + if "nvidia" in lowered or "geforce" in lowered: + return "NVIDIA" + if "intel" in lowered: + return "Intel" + return None + + +def _summary_value(summary: str, key: str) -> str | None: + match = re.search( + rf"(?mi)^\s*{re.escape(key)}\s*=\s*([^\r\n]+?)\s*$", + summary, + ) + return match.group(1).strip()[:255] if match else None + + +def _vulkan_gpu_summaries(summary: str) -> tuple[str, ...]: + headers = tuple(re.finditer(r"(?mi)^\s*GPU\d+\s*:\s*$", summary)) + result: list[str] = [] + for index, header in enumerate(headers): + end = headers[index + 1].start() if index + 1 < len(headers) else len(summary) + block = summary[header.start():end] + device_type = _summary_value(block, "deviceType") + if device_type and "cpu" in device_type.casefold(): + continue + if _summary_value(block, "deviceName"): + result.append(block) + return tuple(result) + + +def _hex_summary_value(summary: str, key: str) -> int | None: + value = _summary_value(summary, key) + if not value: + return None + match = re.search(r"0x([0-9a-f]+)", value, re.IGNORECASE) + if not match: + return None + try: + return int(match.group(1), 16) + except ValueError: + return None + + +def _hex_summary_values(summary: str, key: str) -> set[int]: + values: set[int] = set() + for match in re.finditer( + rf"(?mi)^\s*{re.escape(key)}\s*=\s*0x([0-9a-f]+)\b", + summary, + ): + try: + values.add(int(match.group(1), 16)) + except ValueError: + continue + return values + + +def _gpu_offload_detected(environment: Mapping[str, str]) -> bool: + dri_prime = environment.get("DRI_PRIME", "").casefold().strip() + nvidia_offload = ( + environment.get("__NV_PRIME_RENDER_OFFLOAD", "").casefold().strip() + ) + glx_vendor = environment.get("__GLX_VENDOR_LIBRARY_NAME", "").casefold().strip() + nvidia_optimus = environment.get("__VK_LAYER_NV_optimus", "").strip() + return ( + dri_prime not in {"", "0", "false"} + or nvidia_offload not in {"", "0", "false"} + or glx_vendor == "nvidia" + or bool(nvidia_optimus) + ) + + +def _maximum_version(values: Sequence[str]) -> str | None: + parsed: list[tuple[tuple[int, ...], str]] = [] + for value in values: + match = _VERSION_PATTERN.search(value) + if not match: + continue + version = match.group(0) + try: + parts = tuple(int(part) for part in version.split(".")) + except ValueError: + continue + parsed.append((parts, version)) + if not parsed: + return None + return max(parsed, key=lambda item: item[0])[1] + + +def _run_probe( + argv: Sequence[str], + environment: Mapping[str, str], + timeout_seconds: float, +) -> ProbeResult | None: + if not argv or timeout_seconds <= 0: + return None + probe_environment = os.environ.copy() + probe_environment.update(environment) + probe_environment["LC_ALL"] = "C" + probe_environment.pop("VK_LOADER_DEBUG", None) + try: + completed = subprocess.run( + [str(value) for value in argv], + env=probe_environment, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout_seconds, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + return None + return ProbeResult( + returncode=completed.returncode, + stdout=completed.stdout[-MAX_PROBE_OUTPUT_CHARACTERS:], + stderr=completed.stderr[-MAX_PROBE_OUTPUT_CHARACTERS:], + ) + + +def _probe_text(result: ProbeResult | None) -> str: + if result is None: + return "" + return (result.stdout or result.stderr).strip() + + +def _first_probe_line(result: ProbeResult | None) -> str | None: + text = _probe_text(result) + if not text: + return None + return text.splitlines()[0].strip()[:128] or None + + +def _read_json_object(path: Path) -> dict[str, object] | None: + try: + with path.open("rb") as handle: + content = handle.read(MAX_MARKER_BYTES + 1) + if len(content) > MAX_MARKER_BYTES: + return None + value = json.loads(content.decode("utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + return value if isinstance(value, dict) else None + + +def _object_field( + value: Mapping[str, object] | None, + key: str, +) -> dict[str, object] | None: + if value is None: + return None + field = value.get(key) + return field if isinstance(field, dict) else None + + +def _short_string(value: object) -> str | None: + return value[:255] if isinstance(value, str) and value else None + + +def _optional_bool(value: object) -> bool | None: + return value if isinstance(value, bool) else None + + +def _read_short_text(path: Path) -> str | None: + try: + value = path.read_text(encoding="utf-8", errors="replace").strip() + except OSError: + return None + return value[:255] or None + + +def _read_hex(path: Path) -> int | None: + value = _read_short_text(path) + if value is None: + return None + try: + return int(value, 16) + except ValueError: + return None + + +def _read_decimal(path: Path) -> int | None: + value = _read_short_text(path) + if value is None: + return None + try: + return int(value, 10) + except ValueError: + return None diff --git a/ptlinux/ui/dashboard.py b/ptlinux/ui/dashboard.py index 1a8bf89..5bf5365 100644 --- a/ptlinux/ui/dashboard.py +++ b/ptlinux/ui/dashboard.py @@ -79,6 +79,7 @@ from ..tk_chrome import ( load_square_window_icon, show_themed_dialog, ) +from ..telemetry import current_telemetry, set_shared_telemetry_enabled from .components.action_panel import build_action_panel from .components.broadcasts_panel import build_broadcasts_panel from .components.header import build_dashboard_header @@ -446,6 +447,9 @@ def run_dashboard(paths: AppPaths) -> None: set_root_opacity(1.0) root.lift() play_button.focus_set() + telemetry = current_telemetry() + if telemetry is not None: + telemetry.mark_ready({"interface": "gui"}) root.after_idle(start_remote_content_refresh) def fade_dashboard_in(step: int = 0) -> None: @@ -2832,6 +2836,7 @@ def run_dashboard(paths: AppPaths) -> None: ) def apply_saved_settings(updated_settings: Settings) -> None: + set_shared_telemetry_enabled(updated_settings.telemetry_enabled) saved_limit = ( f"{updated_settings.download_speed_limit_mbps:g} MB/s" if updated_settings.download_speed_limit_mbps is not None @@ -2842,7 +2847,8 @@ def run_dashboard(paths: AppPaths) -> None: f"runtime={updated_settings.runtime}, " f"proton_sync={updated_settings.proton_sync}, " f"download_limit={saved_limit}, " - f"low_bandwidth={updated_settings.low_bandwidth_mode}" + f"low_bandwidth={updated_settings.low_bandwidth_mode}, " + f"telemetry={updated_settings.telemetry_enabled}" ) status.set("Settings saved.") progress_text.set( diff --git a/ptlinux/ui/dialogs/settings_dialog.py b/ptlinux/ui/dialogs/settings_dialog.py index 99ae263..107c20f 100644 --- a/ptlinux/ui/dialogs/settings_dialog.py +++ b/ptlinux/ui/dialogs/settings_dialog.py @@ -180,6 +180,7 @@ def open_settings_dialog( ui_scale_value = tk.StringVar( value=setting_label(current_settings.ui_scale, UI_SCALE_LABELS) ) + telemetry_value = tk.BooleanVar(value=current_settings.telemetry_enabled) suppress_unresponsive_value = tk.BooleanVar( value=current_settings.suppress_unresponsive_warnings ) @@ -313,6 +314,7 @@ def open_settings_dialog( borderless=borderless_value.get(), allow_window_resizing=allow_window_resizing_value.get(), ui_scale=setting_value(ui_scale_value.get(), UI_SCALE_LABELS), + telemetry_enabled=telemetry_value.get(), suppress_unresponsive_warnings=suppress_unresponsive_value.get(), runtime=setting_value(runtime_value.get(), RUNTIME_LABELS), proton_sync=setting_value( @@ -524,13 +526,49 @@ def open_settings_dialog( ) update_custom_download_state() + diagnostics_card = ttk.LabelFrame( + content, + text="PRIVACY & DIAGNOSTICS", + style="SettingsCard.TLabelframe", + padding=14, + ) + diagnostics_card.grid( + row=2, + column=0, + columnspan=2, + sticky="ew", + pady=(12, 0), + ) + diagnostics_card.columnconfigure(0, weight=1) + ttk.Checkbutton( + diagnostics_card, + name="settings_telemetry_enabled", + text="Send anonymous diagnostic and usage information", + variable=telemetry_value, + style="Settings.TCheckbutton", + cursor="hand2", + takefocus=True, + ).grid(row=0, column=0, sticky="w") + diagnostics_hint = ttk.Label( + diagnostics_card, + text=( + "Helps improve PristonTale EU Linux compatibility by sending anonymous " + "launcher errors, Linux and display versions, graphics/runtime compatibility, " + "and installation or update reliability. Game accounts, personal files, " + "hostnames, and hardware serial numbers are never collected." + ), + style="SettingsHint.TLabel", + wraplength=720, + ) + diagnostics_hint.grid(row=1, column=0, sticky="w", pady=(7, 0)) + overrides_card = ttk.LabelFrame( content, text="OPTIONAL OVERRIDES", style="SettingsCard.TLabelframe", padding=14, ) - overrides_card.grid(row=2, column=0, columnspan=2, sticky="ew", pady=(12, 0)) + overrides_card.grid(row=3, column=0, columnspan=2, sticky="ew", pady=(12, 0)) overrides_card.columnconfigure(0, weight=1) ttk.Label( overrides_card, @@ -614,6 +652,9 @@ def open_settings_dialog( card_wraplength = max(140, card_width - round(42 * launcher_ui_scale)) display_hint.configure(wraplength=card_wraplength) rendering_hint.configure(wraplength=card_wraplength) + diagnostics_hint.configure( + wraplength=max(140, viewport_width - round(42 * launcher_ui_scale)) + ) footer_note.configure( wraplength=( max(140, viewport_width - round(42 * launcher_ui_scale)) @@ -644,7 +685,8 @@ def open_settings_dialog( pady=(12, 0), ) downloads_card.grid_configure(row=2, pady=(12, 0)) - overrides_card.grid_configure(row=3, pady=(12, 0)) + diagnostics_card.grid_configure(row=3, pady=(12, 0)) + overrides_card.grid_configure(row=4, pady=(12, 0)) else: display_card.grid_configure( row=0, @@ -663,7 +705,8 @@ def open_settings_dialog( pady=0, ) downloads_card.grid_configure(row=1, pady=(12, 0)) - overrides_card.grid_configure(row=2, pady=(12, 0)) + diagnostics_card.grid_configure(row=2, pady=(12, 0)) + overrides_card.grid_configure(row=3, pady=(12, 0)) if footer_stacked: footer_note.grid_configure(row=0, column=0, columnspan=2) diff --git a/ptlinux/umu.py b/ptlinux/umu.py index 7e29523..4bd070d 100644 --- a/ptlinux/umu.py +++ b/ptlinux/umu.py @@ -13,6 +13,7 @@ import shutil import sys import tarfile import threading +import time from typing import Callable, Mapping, Protocol from urllib.error import HTTPError from urllib.request import Request, urlopen @@ -20,6 +21,11 @@ import zipfile from . import __version__ from .paths import AppPaths +from .telemetry import ( + classify_network_failure, + current_telemetry, + network_http_status, +) UMU_LAUNCHER_VERSION = "1.4.4" @@ -156,11 +162,64 @@ def ensure_managed_umu( archive = paths.managed_umu_archive(UMU_LAUNCHER_VERSION) if not _verified_archive(archive): - _download_archive( - archive, - timeout_seconds=timeout_seconds, - opener=opener, - ) + telemetry = current_telemetry() + download_started_at = time.monotonic() + if telemetry is not None: + telemetry.send_event( + "download_started", + metadata={ + "resource_type": "runtime", + "mirror_id": "github.com", + }, + context={"operation": "runtime_install", "stage": "download"}, + ) + try: + _download_archive( + archive, + timeout_seconds=timeout_seconds, + opener=opener, + ) + except BaseException as exc: + if telemetry is not None: + telemetry.send_event( + "download_failed", + severity="error", + message=str(exc), + metadata={ + "resource_type": "runtime", + "duration_ms": round( + (time.monotonic() - download_started_at) * 1000 + ), + "failure_type": classify_network_failure(exc), + "http_status": network_http_status(exc), + "retry_count": 0, + "resume_used": False, + "mirror_id": "github.com", + }, + context={"operation": "runtime_install", "stage": "download"}, + ) + raise + download_duration = max(0.0, time.monotonic() - download_started_at) + download_bytes = archive.stat().st_size + if telemetry is not None: + telemetry.send_event( + "download_completed", + metadata={ + "resource_type": "runtime", + "download_bytes": download_bytes, + "content_bytes": download_bytes, + "duration_ms": round(download_duration * 1000), + "average_mbps": _average_mbps( + download_bytes, + download_duration, + ), + "retry_count": 0, + "resume_used": False, + "http_status": 200, + "mirror_id": "github.com", + }, + context={"operation": "runtime_install", "stage": "download"}, + ) _install_zipapp(archive, destination) result = _result(destination, changed=True) _write_json( @@ -236,6 +295,12 @@ def _download_archive( raise +def _average_mbps(download_bytes: int, duration_seconds: float) -> float | None: + if download_bytes <= 0 or duration_seconds <= 0: + return None + return round((download_bytes * 8) / duration_seconds / 1_000_000, 3) + + def _install_zipapp(archive: Path, destination: Path) -> None: destination.parent.mkdir(parents=True, exist_ok=True) temporary = _temporary_path(destination, "tmp")