Added better support for kubuntu and fixed some issues with telemetry.

This commit is contained in:
2026-09-09 22:35:13 +02:00
parent 6e6219107e
commit 290d1a5383
8 changed files with 587 additions and 65 deletions

View File

@@ -380,7 +380,9 @@ ptlinux_install_host_dependencies()
ptlinux_detect_missing_dependencies "$family"
if (( ${#PTLINUX_MISSING_DEPENDENCIES[@]} == 0 ))
then
printf '%s\n' "All required host dependencies are already available."
printf '%s\n' \
"All automatically managed host packages are already available." \
"Final graphics and runtime diagnostics run after launcher installation."
return 0
fi
if [[ "$family" == "unknown" ]]

View File

@@ -88,6 +88,7 @@ from .telemetry import (
network_http_status,
shutdown_shared_telemetry,
)
from .system_identity import is_running_inside_steam_runtime
from .umu import (
UmuBootstrapResult,
find_umu_launcher,
@@ -310,12 +311,9 @@ def main(
initialized_telemetry, telemetry_owner = initialize_shared_telemetry(
paths,
session_metadata={"interface": interface, "command": command},
report_launcher_lifecycle=command == "gui",
)
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:
@@ -652,9 +650,24 @@ def _doctor(as_json: bool) -> int:
status = tool.path or "MISSING"
required = "required" if tool.required else "optional"
print(f"{tool.name:12} {status} [{required}] — {tool.purpose}")
optional_missing = [
tool.name for tool in report.tools if not tool.required and tool.path is None
]
if optional_missing:
print(
"Optional tools missing (these do not block launch): "
+ ", ".join(optional_missing)
)
print(f"Host readiness: {'READY' if report.ok else 'BLOCKED'}")
if report.blockers:
print("Blocking diagnostics:")
for blocker in report.blockers:
print(f" - {blocker}")
for warning in report.warnings:
print(f"Warning: {warning}")
if report.package_hint:
if report.package_hint and any(
tool.required and tool.path is None for tool in report.tools
):
print(report.package_hint)
return 0 if report.ok else 1
@@ -1322,6 +1335,28 @@ def _play_locked(
"run pristontale-eu update before playing"
)
settings = _load_or_default_settings(paths, persist=not dry_run)
launcher_nested_in_steam_runtime = is_running_inside_steam_runtime()
if launcher_nested_in_steam_runtime and not dry_run:
print(
"Warning: the launcher itself is running inside Steam Runtime. Start "
"it from the desktop menu or a host terminal; nesting its UMU runtime "
"can cause black game output.",
file=sys.stderr,
flush=True,
)
window_launch_metadata: dict[str, object] = {
"gamebox_enabled": not no_box,
"resizable_window_requested": settings.allow_window_resizing,
"launcher_inside_steam_runtime": launcher_nested_in_steam_runtime,
}
if (
launcher_nested_in_steam_runtime
and settings.allow_window_resizing
and not no_box
):
window_launch_metadata["gamescope_disabled_reason"] = (
"launcher_inside_steam_runtime"
)
start_option = select_depot_start_option(
read_cached_depot_launch_catalog(paths.depot_launch_catalog_cache),
realm or settings.realm,
@@ -1412,11 +1447,13 @@ def _play_locked(
game_started_at = time.monotonic()
launch_success_lock = threading.Lock()
launch_success_reported = False
reported_runtime_warnings: set[str] = set()
if telemetry is not None and not dry_run:
telemetry.send_event(
"game_launch_started",
metadata={
**launch_session_metadata,
**window_launch_metadata,
"launch_duration_ms": round(
(game_started_at - launch_requested_at) * 1000
),
@@ -1438,6 +1475,27 @@ def _play_locked(
"""Timestamp launch success when the supervisor first confirms readiness."""
nonlocal launch_success_reported
presentation_failure = _gamescope_presentation_failure(line)
if telemetry is not None and not dry_run and presentation_failure is not None:
with launch_success_lock:
if presentation_failure not in reported_runtime_warnings:
reported_runtime_warnings.add(presentation_failure)
telemetry.send_event(
"game_window_diagnostic",
severity="warning",
metadata={
**launch_session_metadata,
**window_launch_metadata,
"component": "gamescope_host",
"diagnostic": presentation_failure,
"runtime": selection.kind,
"realm": start_option.name,
},
context={
"operation": "launch_game",
"stage": "presentation",
},
)
window_diagnostic = parse_window_diagnostic(line)
if window_diagnostic is not None:
diagnostic_session_id = window_diagnostic.get("session_id")
@@ -1470,6 +1528,7 @@ def _play_locked(
"game_launch_success",
metadata={
**launch_session_metadata,
**window_launch_metadata,
"launch_duration_ms": round(
(ready_at - launch_requested_at) * 1000
),
@@ -1514,6 +1573,7 @@ def _play_locked(
message="Game launch was interrupted",
metadata={
**launch_session_metadata,
**window_launch_metadata,
"runtime_seconds": round(time.monotonic() - game_started_at),
"clean_exit": False,
"session_duration_class": _session_duration_class(
@@ -1586,6 +1646,7 @@ def _play_locked(
if telemetry is not None and not dry_run:
common_game_metadata: dict[str, object] = {
**launch_session_metadata,
**window_launch_metadata,
"runtime_seconds": round(game_runtime_seconds),
"clean_exit": clean_exit,
"timed_out": outcome.timed_out,
@@ -1610,6 +1671,7 @@ def _play_locked(
"game_launch_success",
metadata={
**launch_session_metadata,
**window_launch_metadata,
"detection_duration_ms": round(game_runtime_seconds * 1000),
"detected_after_exit": True,
"game_window_ready": game_window_ready,
@@ -1896,12 +1958,32 @@ def _runtime_issue_summary(stderr: str) -> dict[str, object]:
"gstreamer_missing_libraries": missing_libraries,
"gstreamer_stream_error": "winegstreamer error" in folded,
"mesa_egl_initialization_failed": "egl: failed to create dri2 screen" in folded,
"gamescope_present_queue_unsupported": (
"physical device queue doesn't support presenting on our surface" in folded
),
"gamescope_vulkan_output_failed_count": folded.count(
"vulkan_make_output failed"
),
"steam_api_app_id_missing": "steamapi_init() failed; no appid found" in folded,
"dbghelp_symdeferred_assertion_count": folded.count(dbghelp_assertion),
"bubblewrap_uid_map_permission_denied": bwrap_uid_map_denied in folded,
}
def _gamescope_presentation_failure(output: str) -> str | None:
folded = output.casefold()
for diagnostic, marker in (
(
"gamescope_present_queue_unsupported",
"physical device queue doesn't support presenting on our surface",
),
("gamescope_vulkan_output_failed", "vulkan_make_output failed"),
):
if marker in folded:
return diagnostic
return None
def _game_window_was_ready(stderr: str) -> bool:
return _game_readiness_signal(stderr) is not None

View File

@@ -42,19 +42,71 @@ class DoctorReport:
package_hint: str | None
warnings: list[str]
@property
def blockers(self) -> tuple[str, ...]:
return host_readiness_blockers(self)
def to_dict(self) -> dict[str, object]:
return asdict(self)
payload = asdict(self)
payload["blockers"] = list(self.blockers)
return payload
def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
def host_readiness_blockers(report: DoctorReport) -> tuple[str, ...]:
"""Explain every condition that makes a host report fail readiness."""
blockers = [
f"{tool.name} is missing ({tool.purpose})."
for tool in report.tools
if tool.required and tool.path is None
]
bubblewrap = next(
(tool for tool in report.tools if tool.name == "bubblewrap"),
None,
)
if (
bubblewrap is not None
and bubblewrap.required
and bubblewrap.path is not None
and not report.container_host.ready
):
issue = report.container_host.issue or "probe failed"
message = f"Bubblewrap cannot create the Steam Runtime sandbox ({issue})."
if report.container_host.recommendation:
message += f" {report.container_host.recommendation}"
blockers.append(message)
if report.container_host.broken_umu_profile_link:
blockers.append(
"UMU's AppArmor profile link is broken; run "
"`pristontale-eu container-host --repair`."
)
if not report.ok and not blockers:
blockers.append(
"The host readiness check failed for an unclassified reason; "
"run `pristontale-eu --json doctor` and report its output."
)
return tuple(blockers)
def inspect_host(
environment: Mapping[str, str] | None = None,
*,
vulkan_support: Mapping[str, object] | None = None,
) -> DoctorReport:
env = os.environ if environment is None else environment
os_release = _read_os_release()
distribution = os_release.get("ID", "unknown")
version = os_release.get("VERSION_ID", "unknown")
architecture = platform.machine()
wine_tools = find_wine_tools()
umu_path = _umu_path(env)
container_host = inspect_container_host(environment=env)
container_required = umu_path is not None
detected_vulkan = (
dict(vulkan_support)
if vulkan_support is not None
else _inspect_vulkan_support(env)
)
tools = [
ToolStatus(
"python3",
@@ -117,9 +169,11 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
"Vulkan diagnostics",
),
]
tools.extend(_platform_optional_tools(distribution, architecture=platform.machine()))
if architecture.casefold() in {"x86_64", "amd64"}:
tools.extend(_vulkan_tool_statuses(detected_vulkan, distribution))
tools.extend(_platform_optional_tools(distribution, architecture=architecture))
warnings: list[str] = []
if platform.machine().casefold() not in {"x86_64", "amd64"}:
if architecture.casefold() not in {"x86_64", "amd64"}:
warnings.append("PristonTale EU requires an x86-64 host with 32-bit Vulkan support.")
if not env.get("WAYLAND_DISPLAY") and not env.get("DISPLAY"):
warnings.append("No graphical X11 or Wayland session is visible.")
@@ -138,7 +192,7 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
"UMU's AppArmor profile link is broken and may prevent profiles from "
"loading after reboot; run `pristontale-eu container-host --repair`."
)
warnings.extend(_arch_runtime_warnings(distribution, platform.machine()))
warnings.extend(_arch_runtime_warnings(distribution, architecture))
required_ok = all(item.path is not None for item in tools if item.required)
return DoctorReport(
ok=(
@@ -148,7 +202,7 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
),
distribution=distribution,
version=version,
architecture=platform.machine(),
architecture=architecture,
display_backend="wayland" if env.get("WAYLAND_DISPLAY") else "x11",
tools=tools,
container_host=container_host,
@@ -228,11 +282,6 @@ def _platform_optional_tools(distribution: str, *, architecture: str) -> list[To
return []
if distribution in ARCH_FAMILY_DISTRIBUTIONS:
return [
_optional_tool_status(
"lib32-vulkan",
"/usr/lib32/libvulkan.so.1",
"32-bit Vulkan runtime compatibility layer",
),
_optional_tool_status(
"lib32-gstreamer",
"/usr/lib32/gstreamer-1.0",
@@ -250,11 +299,6 @@ def _arch_runtime_warnings(distribution: str, architecture: str) -> list[str]:
"Non-x86_64 Linux hosts are not a supported runtime architecture for this game."
]
warnings: list[str] = []
if not _path_exists("/usr/lib32/libvulkan.so.1"):
warnings.append(
"Install lib32-vulkan-icd-loader and the GPU vendor's matching 32-bit "
"Vulkan driver; this 32-bit D3D9 client otherwise cannot initialize DXVK."
)
if not _path_exists("/usr/lib32/gstreamer-1.0"):
warnings.append(
"Install lib32-gstreamer and lib32-gst-plugins-base if the login video or "
@@ -264,6 +308,77 @@ def _arch_runtime_warnings(distribution: str, architecture: str) -> list[str]:
return warnings
def _inspect_vulkan_support(environment: Mapping[str, str]) -> dict[str, object]:
try:
from .telemetry_environment import inspect_vulkan_support
return inspect_vulkan_support(environment)
except Exception:
return {}
def _vulkan_tool_statuses(
support: Mapping[str, object],
distribution: str,
) -> list[ToolStatus]:
return [
_vulkan_tool_status(64, support, distribution),
_vulkan_tool_status(32, support, distribution),
]
def _vulkan_tool_status(
bits: int,
support: Mapping[str, object],
distribution: str,
) -> ToolStatus:
available = support.get(f"vulkan_{bits}bit") is True
loader_available = support.get(f"vulkan_{bits}bit_loader_available") is True
icd_available = support.get(f"vulkan_{bits}bit_icd_available") is True
purpose = f"{bits}-bit Vulkan loader and GPU ICD required by DXVK"
if not available:
missing_parts: list[str] = []
if not loader_available:
missing_parts.append("loader")
if not icd_available:
missing_parts.append("GPU driver/ICD")
purpose = (
f"{bits}-bit Vulkan {' and '.join(missing_parts)} required by DXVK; "
+ _vulkan_package_guidance(distribution, bits)
)
return ToolStatus(
f"vulkan-{bits}bit",
"loader+ICD" if available else None,
True,
purpose,
)
def _vulkan_package_guidance(distribution: str, bits: int) -> str:
suffix = ":i386" if bits == 32 else ""
if distribution == "ubuntu":
return (
"on Ubuntu/Kubuntu install "
f"libvulkan1{suffix} and the matching GPU package "
f"(mesa-vulkan-drivers{suffix} or libnvidia-gl-<version>{suffix})"
)
if distribution in {"debian", "linuxmint", "pop", "elementary", "zorin"}:
return (
f"with APT install libvulkan1{suffix} and the matching {bits}-bit GPU ICD"
)
if distribution in {"fedora", "nobara"}:
return f"install the matching {bits}-bit Vulkan loader and GPU driver with DNF"
if distribution in ARCH_FAMILY_DISTRIBUTIONS:
return (
"install vulkan-icd-loader and the GPU vendor driver"
if bits == 64
else "install lib32-vulkan-icd-loader and the matching lib32 GPU driver"
)
if distribution in {"opensuse-tumbleweed", "opensuse-leap", "suse"}:
return f"install the matching {bits}-bit Vulkan loader and GPU driver with Zypper"
return f"install the matching {bits}-bit Vulkan loader and GPU driver"
def _optional_tool_status(name: str, path: str, purpose: str) -> ToolStatus:
if _path_exists(path):
return ToolStatus(name, path, required=False, purpose=purpose)

View File

@@ -479,6 +479,18 @@ class _X11:
return None
return values[0]
def restore_normal_window_state(self, window: int) -> bool:
"""Undo a nested game iconify without activating the outer host window."""
self._replace_long_property(
window,
b"WM_STATE",
b"WM_STATE",
(_X_WM_STATE_NORMAL, 0),
)
self.library.XSync(self.display, 0)
return self.window_state(window) == _X_WM_STATE_NORMAL
def window_is_borderless(self, window: int) -> bool | None:
"""Return whether the client currently requests no Motif decorations."""
@@ -1776,9 +1788,38 @@ def _resolve_gamescope_game_window(
wm_state=_wm_state_name(state),
)
if state == _X_WM_STATE_ICONIC:
# IconicState can be a real desktop minimize request. Monitoring must never
# turn that user-controlled state back into a visible window.
return None
# This mapper runs inside Gamescope's nested X server. The desktop owns a
# separate outer window, so restoring this game-driven login transition
# neither activates nor raises a minimized host window.
try:
restored = mapper.restore_normal_window_state(window)
except Exception as exc:
restored = False
if diagnostics is not None:
diagnostics.record(
"nested_iconify_recovery_failed",
window_id=window,
error_type=type(exc).__name__,
)
if restored:
if diagnostics is not None:
diagnostics.record(
"nested_iconify_recovered",
window_id=window,
**_sample_window_diagnostic_details(
diagnostics,
mapper,
window,
),
)
print(
"Game login window requested nested minimization; kept its "
"surface visible inside Gamescope.",
file=sys.stderr,
flush=True,
)
return window
state = None
if state is not None and state != _X_WM_STATE_WITHDRAWN:
return window

View File

@@ -428,10 +428,41 @@ def _bundled_gamescope_is_usable() -> bool:
return os.access(path, os.X_OK)
def _select_wayland_gamescope(executable: str) -> tuple[str, bool, bool]:
def _is_kde_session(environment: Mapping[str, str]) -> bool:
"""Recognize Plasma sessions without relying on an Ubuntu flavor name."""
if environment.get("KDE_FULL_SESSION", "").strip().casefold() in {
"1",
"true",
"yes",
}:
return True
if environment.get("KDE_SESSION_VERSION", "").strip():
return True
for key in ("XDG_CURRENT_DESKTOP", "XDG_SESSION_DESKTOP", "DESKTOP_SESSION"):
tokens = re.split(
r"[^a-z0-9]+",
environment.get(key, "").strip().casefold(),
)
if "kde" in tokens or any(token.startswith("plasma") for token in tokens):
return True
return False
def _select_wayland_gamescope(
executable: str,
*,
environment: Mapping[str, str] | None = None,
) -> tuple[str, bool, bool]:
"""Choose a direct-safe Gamescope or request the stable composed path."""
version = _read_gamescope_version(executable)
env = os.environ if environment is None else environment
if _is_kde_session(env):
# This is reached only when Plasma has no usable XWayland display. Keep
# native Wayland fully composed because newer Gamescope builds still have
# KWin-specific presentation failures outside our backing-surface fixes.
return executable, True, False
if version is not None and version >= WAYLAND_BACKING_FIX_VERSION:
return executable, False, False
if version == BUNDLED_GAMESCOPE_VERSION and _bundled_gamescope_is_usable():
@@ -458,9 +489,15 @@ def _wayland_socket_available(environment: Mapping[str, str]) -> bool:
def _select_host_backend(environment: Mapping[str, str]) -> str:
"""Prefer Wayland's retained-buffer resize path when it is usable."""
"""Choose the host backend with a Plasma-safe XWayland fallback."""
if _wayland_socket_available(environment):
# Gamescope's native Wayland backend still has KWin-specific presentation,
# fullscreen, and fractional-scaling failures. Plasma normally exposes an
# XWayland display, so keep the resizable host window on the mature SDL/X11
# path while leaving non-Plasma Wayland sessions on direct presentation.
if _is_kde_session(environment) and environment.get("DISPLAY", "").strip():
return "sdl"
return "wayland"
return "sdl"
@@ -629,6 +666,13 @@ def run_gamescope(command: Sequence[str], width: int, height: int) -> int:
diagnostics = WindowDiagnosticRecorder("gamescope_host")
environment = os.environ.copy()
backend = _select_host_backend(environment)
kde_session = _is_kde_session(environment)
kde_wayland_xwayland_fallback = (
backend == "sdl"
and kde_session
and bool(environment.get("WAYLAND_DISPLAY", "").strip())
and bool(environment.get("DISPLAY", "").strip())
)
selected_command = list(map(str, command))
force_composition = backend == "sdl"
using_bundled_gamescope = False
@@ -637,7 +681,10 @@ def run_gamescope(command: Sequence[str], width: int, height: int) -> int:
selected_command[0],
force_composition,
using_bundled_gamescope,
) = _select_wayland_gamescope(selected_command[0])
) = _select_wayland_gamescope(
selected_command[0],
environment=environment,
)
mapper = _open_host_x11() if backend == "sdl" else None
diagnostics.record(
"runtime_configured",
@@ -647,6 +694,22 @@ def run_gamescope(command: Sequence[str], width: int, height: int) -> int:
display_available=bool(environment.get("DISPLAY")),
wayland_display_available=bool(environment.get("WAYLAND_DISPLAY")),
host_x11_monitor_available=mapper is not None,
kde_session=kde_session,
kde_wayland_compatibility_mode=(
kde_wayland_xwayland_fallback
or (backend == "wayland" and kde_session and force_composition)
),
kde_wayland_xwayland_fallback=kde_wayland_xwayland_fallback,
backend_reason=(
"kde_wayland_xwayland_compatibility"
if kde_wayland_xwayland_fallback
else "wayland_socket"
if backend == "wayland"
else "x11_or_unavailable_wayland"
),
presentation_mode=(
f"{backend}_{'composed' if force_composition else 'direct'}"
),
canvas_width=width,
canvas_height=height,
)
@@ -740,6 +803,14 @@ def run_gamescope(command: Sequence[str], width: int, height: int) -> int:
diagnostics.close(process_started=False, backend=backend)
raise GamescopeRuntimeError(f"could not start Gamescope: {exc}") from exc
if kde_wayland_xwayland_fallback:
print(
"KDE Wayland compatibility mode active: Gamescope is using its "
"SDL host window through XWayland to avoid KWin presentation, "
"fullscreen, and fractional-scaling failures.",
file=sys.stderr,
flush=True,
)
if backend == "wayland" and not force_composition:
message = (
"Gamescope resize optimization active: native Wayland direct "
@@ -751,6 +822,13 @@ def run_gamescope(command: Sequence[str], width: int, height: int) -> int:
"decoration-input, and login-window visibility fixes are active."
)
print(message, file=sys.stderr, flush=True)
elif backend == "wayland" and kde_session:
print(
"KDE Wayland compatibility mode active: Gamescope is using its "
"composed native fallback because no XWayland display is available.",
file=sys.stderr,
flush=True,
)
elif backend == "wayland":
print(
"Warning: this Gamescope build lacks safe direct Wayland letterbox "

View File

@@ -26,6 +26,10 @@ from .gamebox import (
)
from .paths import AppPaths
from .settings import Settings
from .system_identity import (
detect_distribution_id as _detect_distribution_id,
is_running_inside_steam_runtime as _is_running_inside_steam_runtime,
)
from .umu import find_umu_launcher
@@ -468,7 +472,14 @@ def run_game(
if use_gamebox:
gamescope_path = None
game_size = None
if settings.allow_window_resizing:
if settings.allow_window_resizing and launcher_inside_steam_runtime():
print(
"Warning: resizable Gamescope was disabled because the launcher "
"is already inside Steam Runtime; using a fixed game window.",
file=sys.stderr,
flush=True,
)
elif settings.allow_window_resizing:
gamescope_environment = os.environ.copy()
gamescope_environment.update(environment)
gamescope_path = find_gamescope(gamescope_environment)
@@ -714,11 +725,13 @@ def _proton_environment(
def detect_distribution_id() -> str:
"""Return the freedesktop distribution ID without invoking distro tools."""
try:
value = platform.freedesktop_os_release().get("ID", "")
except OSError:
return ""
return value.strip().casefold()
return _detect_distribution_id()
def launcher_inside_steam_runtime() -> bool:
"""Return whether this launcher process is nested in Steam Runtime."""
return _is_running_inside_steam_runtime()
def is_cachyos_host(

View File

@@ -27,6 +27,7 @@ import uuid
from . import __version__
from .paths import AppPaths
from .system_identity import SystemIdentity, detect_system_identity, read_os_release
from .telemetry_environment import collect_extended_environment
@@ -267,6 +268,7 @@ class TelemetryService:
environment: Mapping[str, str] | None = None,
transport: TelemetryTransport | None = None,
session_metadata: Mapping[str, object] | None = None,
report_launcher_lifecycle: bool | None = None,
monotonic: Callable[[], float] = time.monotonic,
) -> None:
self.paths = paths
@@ -283,10 +285,15 @@ class TelemetryService:
self._session_metadata = dict(session_metadata or {})
self._monotonic = monotonic
self._started_at = monotonic()
self._reports_launcher_lifecycle = (
process_reports_launcher_lifecycle = (
self._environment.get(TELEMETRY_PROCESS_ROLE_VARIABLE, "").strip().casefold()
!= TELEMETRY_OPERATION_PROCESS_ROLE
)
self._reports_launcher_lifecycle = process_reports_launcher_lifecycle and (
True
if report_launcher_lifecycle is None
else bool(report_launcher_lifecycle)
)
self._enabled = bool(enabled)
self._active = False
self._accepting = False
@@ -923,6 +930,7 @@ class TelemetryService:
def _collect_common_payload(self, first_seen: str) -> dict[str, object]:
environment = self._environment
system_identity = detect_system_identity()
payload: dict[str, object] = {
"launcher_version": __version__,
"launcher_build": environment.get("PTLINUX_BUILD", __version__)[:32],
@@ -934,10 +942,13 @@ class TelemetryService:
"install_is_external": _install_is_external(self.paths, environment),
"configuration_schema_version": _settings_schema_version(),
},
"os": _operating_system_info(),
"os": _operating_system_info(system_identity),
"display": _display_info(environment),
"filesystem": _filesystem_info(self.paths.data),
}
execution_environment = _execution_environment_info(system_identity)
if execution_environment is not None:
payload["execution_environment"] = execution_environment
commit = environment.get("PTLINUX_COMMIT")
if commit and re.fullmatch(r"[0-9a-fA-F]{7,40}", commit):
payload["launcher_commit"] = commit[:12].lower()
@@ -1097,6 +1108,7 @@ def initialize_shared_telemetry(
paths: AppPaths,
*,
session_metadata: Mapping[str, object] | None = None,
report_launcher_lifecycle: bool | None = None,
) -> tuple[TelemetryService, bool]:
"""Return the process-wide service and whether this caller owns its shutdown."""
@@ -1108,6 +1120,7 @@ def initialize_shared_telemetry(
paths,
enabled=_telemetry_enabled_from_settings(paths),
session_metadata=session_metadata,
report_launcher_lifecycle=report_launcher_lifecycle,
)
_shared_service = service
service.initialize()
@@ -1230,8 +1243,11 @@ def _fsync_directory(path: Path) -> None:
os.close(descriptor)
def _operating_system_info() -> dict[str, object]:
release = _read_os_release()
def _operating_system_info(
identity: SystemIdentity | None = None,
) -> dict[str, object]:
identity = detect_system_identity() if identity is None else identity
release = identity.operating_system_release
return {
"name": release.get("NAME", platform.system() or "unknown"),
"id": release.get("ID", "unknown"),
@@ -1241,27 +1257,38 @@ def _operating_system_info() -> dict[str, object]:
"architecture": platform.machine() or "unknown",
"machine_architecture": platform.machine() or "unknown",
"glibc_version": platform.libc_ver()[1] or None,
"identity_source": (
"container_host_os_release"
if identity.host_release is not None
else "execution_os_release"
),
}
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
return read_os_release(path)
def _execution_environment_info(
identity: SystemIdentity,
) -> dict[str, object] | None:
if not identity.running_inside_steam_runtime:
return None
release = identity.execution_release
return {
"containerized": True,
"kind": "steam_runtime",
"id": release.get("ID", "steamrt"),
"name": release.get("NAME", "Steam Runtime"),
"version": release.get(
"PRETTY_NAME",
release.get("VERSION", "unknown"),
),
"version_id": release.get("VERSION_ID", "unknown"),
"host_os_release_available": identity.host_release is not None,
"launcher_inside_game_runtime": True,
"glibc_version": platform.libc_ver()[1] or None,
}
def _display_info(environment: Mapping[str, str]) -> dict[str, object]:

View File

@@ -106,6 +106,7 @@ def collect_extended_environment(
"dependencies": dependencies,
}
display = _display_info(sysfs_drm)
display.update(_kde_display_scaling_info(environment, run_probe))
if display:
result["display"] = display
if gpu:
@@ -263,6 +264,23 @@ def _vulkan_info(
return result
def inspect_vulkan_support(
environment: Mapping[str, str],
*,
probe_runner: ProbeRunner | None = None,
manifest_directories: Sequence[Path] = _VULKAN_MANIFEST_DIRECTORIES,
library_roots: Sequence[Path] = _LIBRARY_ROOTS,
) -> dict[str, object]:
"""Return host Vulkan loader and ICD readiness for dependency checks."""
return _vulkan_info(
environment,
probe_runner or _run_probe,
manifest_directories=manifest_directories,
library_roots=library_roots,
)
def _gpu_info(
vulkan_summary: object,
*,
@@ -362,7 +380,7 @@ def _dependency_health(
try:
from .doctor import inspect_host
report = inspect_host(environment)
report = inspect_host(environment, vulkan_support=runtime)
for item in report.tools:
available = item.path is not None
tools[item.name] = available
@@ -834,15 +852,18 @@ def _vulkan_manifests(
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
override = environment.get("VK_DRIVER_FILES", "").strip()
if not override:
override = environment.get("VK_ICD_FILENAMES", "").strip()
if override:
candidates.extend(_manifest_entries(override))
else:
candidates.extend(_manifest_entries(environment.get("VK_ADD_DRIVER_FILES", "")))
for directory in _vulkan_manifest_search_directories(environment, directories):
try:
candidates.extend(sorted(directory.glob("*.json")))
except OSError:
continue
seen: set[Path] = set()
result: list[Path] = []
for candidate in candidates:
@@ -857,6 +878,54 @@ def _vulkan_manifests(
return tuple(result)
def _manifest_entries(value: str) -> tuple[Path, ...]:
result: list[Path] = []
for raw_entry in value.split(os.pathsep):
if not raw_entry:
continue
entry = Path(raw_entry).expanduser()
try:
if entry.is_dir():
result.extend(sorted(entry.glob("*.json")))
else:
result.append(entry)
except OSError:
continue
return tuple(result)
def _vulkan_manifest_search_directories(
environment: Mapping[str, str],
directories: Sequence[Path],
) -> tuple[Path, ...]:
if directories is not _VULKAN_MANIFEST_DIRECTORIES:
return tuple(dict.fromkeys(directories))
home = environment.get("HOME", "").strip()
config_home = environment.get("XDG_CONFIG_HOME", "").strip()
data_home = environment.get("XDG_DATA_HOME", "").strip()
config_directories = environment.get("XDG_CONFIG_DIRS", "/etc/xdg")
data_directories = environment.get(
"XDG_DATA_DIRS",
"/usr/local/share:/usr/share",
)
result = list(directories)
if config_home:
result.append(Path(config_home).expanduser() / "vulkan/icd.d")
elif home:
result.append(Path(home).expanduser() / ".config/vulkan/icd.d")
for value in config_directories.split(os.pathsep):
if value:
result.append(Path(value).expanduser() / "vulkan/icd.d")
if data_home:
result.append(Path(data_home).expanduser() / "vulkan/icd.d")
elif home:
result.append(Path(home).expanduser() / ".local/share/vulkan/icd.d")
for value in data_directories.split(os.pathsep):
if value:
result.append(Path(value).expanduser() / "vulkan/icd.d")
return tuple(dict.fromkeys(result))
def _vulkan_loader_architectures(library_roots: Sequence[Path]) -> set[int]:
architectures: set[int] = set()
for root in library_roots:
@@ -967,6 +1036,101 @@ def _display_info(sysfs_drm: Path) -> dict[str, object]:
return {"monitor_count": monitor_count} if monitor_count is not None else {}
def _kde_display_scaling_info(
environment: Mapping[str, str],
probe_runner: ProbeRunner,
) -> dict[str, object]:
"""Collect bounded Plasma scale facts without monitor names or resolutions."""
if not environment.get("WAYLAND_DISPLAY", "").strip() or not _is_kde_desktop(
environment
):
return {}
result: dict[str, object] = {"kde_wayland": True}
executable = shutil.which(
"kscreen-doctor",
path=environment.get("PATH", ""),
)
result["kscreen_doctor_available"] = executable is not None
if executable is None:
return result
probe = probe_runner(
(executable, "--json"),
environment,
PROBE_TIMEOUT_SECONDS,
)
if probe is None or probe.returncode != 0:
result["scale_probe_succeeded"] = False
return result
try:
payload = json.loads(probe.stdout)
except (json.JSONDecodeError, TypeError, UnicodeError):
result["scale_probe_succeeded"] = False
return result
if not isinstance(payload, dict):
result["scale_probe_succeeded"] = False
return result
raw_outputs = payload.get("outputs")
if isinstance(raw_outputs, list):
outputs = raw_outputs
elif isinstance(raw_outputs, dict):
outputs = list(raw_outputs.values())
else:
outputs = []
scales: list[float] = []
for output in outputs[:32]:
if not isinstance(output, dict):
continue
if output.get("connected") is False or output.get("enabled") is False:
continue
value = output.get("scale")
if isinstance(value, bool):
continue
try:
scale = float(value)
except (TypeError, ValueError):
continue
if not 0.5 <= scale <= 8.0:
continue
scales.append(round(scale, 3))
result["scale_probe_succeeded"] = bool(scales)
if not scales:
return result
unique_scales = sorted(set(scales))
result.update(
{
"scale_output_count": len(scales),
"scale_min": unique_scales[0],
"scale_max": unique_scales[-1],
"fractional_scaling": any(
abs(scale - round(scale)) > 0.001 for scale in unique_scales
),
"mixed_scaling": len(unique_scales) > 1,
"scale_detection": "kscreen_doctor",
}
)
return result
def _is_kde_desktop(environment: Mapping[str, str]) -> bool:
if environment.get("KDE_FULL_SESSION", "").strip().casefold() in {
"1",
"true",
"yes",
}:
return True
if environment.get("KDE_SESSION_VERSION", "").strip():
return True
for key in ("XDG_CURRENT_DESKTOP", "XDG_SESSION_DESKTOP", "DESKTOP_SESSION"):
tokens = re.split(
r"[^a-z0-9]+",
environment.get(key, "").strip().casefold(),
)
if "kde" in tokens or any(token.startswith("plasma") for token in tokens):
return True
return False
def _connected_monitor_count(sysfs_drm: Path) -> int | None:
try:
connectors = sorted(sysfs_drm.glob("card[0-9]*-*/status"))[:32]