Reworked UI for easier workflow.

2026-09-03 19:07:09 +02:00
parent c61254e4a9
commit ea323e3ab0
21 changed files with 3300 additions and 2547 deletions
--- a/ptlinux/ui/viewmodels/init.py
+++ b/ptlinux/ui/viewmodels/init.py
@@ -0,0 +1 @@
+"""Pure state and presentation models used by the native dashboard."""
--- a/ptlinux/ui/viewmodels/dashboard.py
+++ b/ptlinux/ui/viewmodels/dashboard.py
@@ -0,0 +1,480 @@
+"""Dashboard state and presentation helpers independent of Tk widget layout."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import math
+import os
+import signal
+import subprocess
+import sys
+import threading
+import time
+from typing import TYPE_CHECKING, Callable, Mapping
+
+from ...paths import AppPaths
+from ...self_update import LauncherUpdateCheck
+
+
+if TYPE_CHECKING:
+	from tkinter import Frame
+
+
+WINDOW_WIDTH = 1002
+WINDOW_HEIGHT = 784
+BASE_DISPLAY_DPI = 96.0
+BASE_TK_SCALING = BASE_DISPLAY_DPI / 72.0
+UI_SCALE_FACTORS: dict[str, float] = {
+	"100": 1.0,
+	"125": 1.25,
+	"150": 1.5,
+	"175": 1.75,
+	"200": 2.0,
+}
+UI_SCALE_LABELS: dict[str, str] = {
+	"Automatic (system DPI)": "auto",
+	"100% — Standard": "100",
+	"125% — Large": "125",
+	"150% — Larger": "150",
+	"175% — Extra large": "175",
+	"200% — Maximum": "200",
+}
+NEWS_THREE_COLUMN_WIDTH = 900
+NEWS_TWO_COLUMN_WIDTH = 620
+STACKED_CONTROLS_WIDTH = 860
+POLL_INTERVAL_MILLISECONDS = 33
+ACTIVE_POLL_INTERVAL_MILLISECONDS = 75
+IDLE_POLL_INTERVAL_MILLISECONDS = 500
+DASHBOARD_PRELOADER_MINIMUM_MILLISECONDS = 300
+DASHBOARD_PRELOADER_TIMEOUT_MILLISECONDS = 10_000
+DASHBOARD_TRANSITION_STEP_MILLISECONDS = 16
+DASHBOARD_TRANSITION_STEPS = 9
+DASHBOARD_TRANSITION_MINIMUM_OPACITY = 0.18
+STARTUP_PROGRESS_ANIMATION_INTERVAL_MILLISECONDS = 16
+STARTUP_PROGRESS_ANIMATION_EASING = 0.28
+STARTUP_PROGRESS_ANIMATION_MINIMUM_STEP = 0.08
+PRELOADER_STATUS_CHARACTERS = 64
+DASHBOARD_STARTUP_COMPONENTS = frozenset(
+	{
+		"broadcast",
+		"cached_content",
+		"compatibility",
+		"launcher_update",
+		"logo",
+		"news",
+		"news_artwork",
+		"patches",
+		"server_status",
+		"version",
+	}
+)
+DASHBOARD_STARTUP_REQUIRED_COMPONENTS = frozenset({"cached_content"})
+
+
+@dataclass(frozen=True)
+class LauncherUpdatePresentation:
+	"""Visual treatment for one authenticated launcher-update result."""
+
+	button_text: str
+	button_style: str
+	badge_text: str | None
+
+
+@dataclass(frozen=True)
+class GameClientDebugState:
+	"""Live process state exposed only in the hidden dashboard debug view."""
+
+	client_id: int
+	phase: str
+	supervisor_pid: int | None
+	elapsed_seconds: float
+
+
+class DashboardError(RuntimeError):
+	"""Raised when the native dashboard cannot be displayed."""
+
+
+def resolve_ui_scale(preference: str, pixels_per_inch: float) -> float:
+	"""Resolve a saved launcher scale while respecting the detected display DPI."""
+
+	if preference == "auto":
+		if not math.isfinite(pixels_per_inch) or pixels_per_inch <= 0:
+			return 1.0
+		return min(2.0, max(1.0, pixels_per_inch / BASE_DISPLAY_DPI))
+	try:
+		return UI_SCALE_FACTORS[preference]
+	except KeyError as error:
+		raise ValueError(f"unsupported launcher UI scale: {preference}") from error
+
+
+def fitted_window_size(
+	width: int,
+	height: int,
+	*,
+	scale: float,
+	screen_width: int,
+	screen_height: int,
+	margin: int = 64,
+) -> tuple[int, int]:
+	"""Scale an initial window without placing it beyond the available display."""
+
+	if width <= 0 or height <= 0 or scale <= 0:
+		raise ValueError("window dimensions and scale must be positive")
+	if screen_width <= 0 or screen_height <= 0 or margin < 0:
+		raise ValueError("screen dimensions must be positive and margin non-negative")
+	available_width = max(1, screen_width - margin)
+	available_height = max(1, screen_height - margin)
+	return (
+		min(max(1, round(width * scale)), available_width),
+		min(max(1, round(height * scale)), available_height),
+	)
+
+
+def news_card_columns(width: int, scale: float) -> int:
+	"""Choose a readable card grid for the current physical dashboard width."""
+
+	if width <= 0 or scale <= 0:
+		return 1
+	logical_width = width / scale
+	if logical_width >= NEWS_THREE_COLUMN_WIDTH:
+		return 3
+	if logical_width >= NEWS_TWO_COLUMN_WIDTH:
+		return 2
+	return 1
+
+
+def dashboard_controls_stacked(width: int, scale: float) -> bool:
+	return width <= 0 or scale <= 0 or width / scale < STACKED_CONTROLS_WIDTH
+
+
+def setting_label(value: str, choices: Mapping[str, str]) -> str:
+	for label, choice in choices.items():
+		if choice == value:
+			return label
+	raise ValueError(f"unsupported saved setting value: {value}")
+
+
+def setting_value(label: str, choices: Mapping[str, str]) -> str:
+	try:
+		return choices[label]
+	except KeyError as error:
+		raise ValueError(f"unsupported setting choice: {label}") from error
+
+
+class GameLaunchSupervisorRegistry:
+	"""Bridge Tk cancellation requests to game launch worker processes safely."""
+
+	def __init__(self) -> None:
+		self._lock = threading.Lock()
+		self._processes: dict[int, subprocess.Popen[str]] = {}
+		self._cancel_requested: set[int] = set()
+
+	def attach(self, client_id: int, process: subprocess.Popen[str]) -> bool:
+		"""Register a supervisor and report whether cancellation raced its start."""
+
+		with self._lock:
+			self._processes[client_id] = process
+			return client_id in self._cancel_requested
+
+	def request_cancel(self, client_id: int) -> subprocess.Popen[str] | None:
+		"""Record cancellation and return an already-started supervisor, if any."""
+
+		with self._lock:
+			self._cancel_requested.add(client_id)
+			return self._processes.get(client_id)
+
+	def detach(self, client_id: int, process: subprocess.Popen[str]) -> None:
+		with self._lock:
+			if self._processes.get(client_id) is process:
+				self._processes.pop(client_id, None)
+
+	def is_cancel_requested(self, client_id: int) -> bool:
+		with self._lock:
+			return client_id in self._cancel_requested
+
+	def finish(self, client_id: int) -> bool:
+		"""Forget one launch and return whether the user cancelled it."""
+
+		with self._lock:
+			self._processes.pop(client_id, None)
+			if client_id not in self._cancel_requested:
+				return False
+			self._cancel_requested.remove(client_id)
+			return True
+
+
+def interrupt_game_supervisor(process: subprocess.Popen[str]) -> bool:
+	"""Ask the play CLI to stop its complete Proton/game process group."""
+
+	if process.poll() is not None:
+		return False
+	try:
+		process.send_signal(signal.SIGINT)
+	except OSError:
+		return False
+	return True
+
+
+class GameClientTracker:
+	"""Track dashboard-launched clients without retaining process objects in Tk."""
+
+	def __init__(self, *, clock: Callable[[], float] = time.monotonic) -> None:
+		self._clock = clock
+		self._next_client_id = 1
+		self._starting: set[int] = set()
+		self._running: set[int] = set()
+		self._started_at: dict[int, float] = {}
+		self._running_at: dict[int, float] = {}
+		self._supervisor_pids: dict[int, int] = {}
+
+	def begin(self) -> int:
+		client_id = self._next_client_id
+		self._next_client_id += 1
+		self._starting.add(client_id)
+		self._started_at[client_id] = self._clock()
+		return client_id
+
+	def attach_supervisor(self, client_id: int, pid: int) -> bool:
+		if pid <= 0 or (
+			client_id not in self._starting and client_id not in self._running
+		):
+			return False
+		self._supervisor_pids[client_id] = pid
+		return True
+
+	def mark_running(self, client_id: int) -> bool:
+		if client_id in self._running or client_id not in self._starting:
+			return False
+		self._starting.remove(client_id)
+		self._running.add(client_id)
+		self._running_at[client_id] = self._clock()
+		return True
+
+	def finish(self, client_id: int) -> str | None:
+		if client_id in self._starting:
+			self._starting.remove(client_id)
+			phase = "starting"
+		elif client_id in self._running:
+			self._running.remove(client_id)
+			phase = "running"
+		else:
+			return None
+		self._started_at.pop(client_id, None)
+		self._running_at.pop(client_id, None)
+		self._supervisor_pids.pop(client_id, None)
+		return phase
+
+	def debug_states(
+		self,
+		*,
+		timestamp: float | None = None,
+	) -> tuple[GameClientDebugState, ...]:
+		now = self._clock() if timestamp is None else timestamp
+		return tuple(
+			GameClientDebugState(
+				client_id=client_id,
+				phase="running" if client_id in self._running else "starting",
+				supervisor_pid=self._supervisor_pids.get(client_id),
+				elapsed_seconds=max(
+					0.0,
+					now
+					- (
+						self._running_at.get(client_id, now)
+						if client_id in self._running
+						else self._started_at.get(client_id, now)
+					),
+				),
+			)
+			for client_id in sorted(self._starting | self._running)
+		)
+
+	@property
+	def starting_count(self) -> int:
+		return len(self._starting)
+
+	@property
+	def running_count(self) -> int:
+		return len(self._running)
+
+	@property
+	def active_count(self) -> int:
+		return self.starting_count + self.running_count
+
+
+def game_client_activity_text(running_count: int) -> str:
+	if running_count < 0:
+		raise ValueError("running client count must not be negative")
+	if running_count == 0:
+		return ""
+	label = "client" if running_count == 1 else "clients"
+	return f"{running_count} {label} running"
+
+
+def debug_elapsed_text(elapsed_seconds: float) -> str:
+	"""Format a live debug duration without sub-second UI churn."""
+
+	seconds = max(0, int(elapsed_seconds))
+	hours, remainder = divmod(seconds, 60 * 60)
+	minutes, seconds = divmod(remainder, 60)
+	if hours:
+		return f"{hours}h {minutes}m {seconds}s"
+	if minutes:
+		return f"{minutes}m {seconds}s"
+	return f"{seconds}s"
+
+
+def launcher_update_presentation(
+	check: LauncherUpdateCheck,
+) -> LauncherUpdatePresentation:
+	"""Make authenticated updates visually distinct from the normal version button."""
+
+	if check.available and check.supported:
+		return LauncherUpdatePresentation(
+			button_text=f"UPDATE LAUNCHER · {check.latest_version}",
+			button_style="LauncherUpdate.TButton",
+			badge_text="SIGNED UPDATE AVAILABLE",
+		)
+	if check.available:
+		return LauncherUpdatePresentation(
+			button_text=f"MANUAL UPDATE · {check.latest_version}",
+			button_style="LauncherManualUpdate.TButton",
+			badge_text="MANUAL UPDATE REQUIRED",
+		)
+	return LauncherUpdatePresentation(
+		button_text=f"LAUNCHER {check.current_version}",
+		button_style="Secondary.TButton",
+		badge_text=None,
+	)
+
+
+def dashboard_preloader_delay(
+	started_at: float,
+	*,
+	timestamp: float | None = None,
+) -> int:
+	"""Keep the startup preloader visible long enough to avoid a brief flash."""
+
+	now = time.monotonic() if timestamp is None else timestamp
+	elapsed_milliseconds = round(max(0.0, now - started_at) * 1000)
+	return max(0, DASHBOARD_PRELOADER_MINIMUM_MILLISECONDS - elapsed_milliseconds)
+
+
+def dashboard_content_ready(ready_components: set[str] | frozenset[str]) -> bool:
+	"""Return whether cached content is ready for an interactive dashboard."""
+
+	return DASHBOARD_STARTUP_REQUIRED_COMPONENTS.issubset(ready_components)
+
+
+def dashboard_transition_opacity(step: int, *, fading_in: bool) -> float:
+	"""Return one clamped opacity step for the loader/dashboard cross-fade."""
+
+	progress = min(max(step, 0), DASHBOARD_TRANSITION_STEPS) / DASHBOARD_TRANSITION_STEPS
+	opacity_range = 1.0 - DASHBOARD_TRANSITION_MINIMUM_OPACITY
+	if fading_in:
+		return DASHBOARD_TRANSITION_MINIMUM_OPACITY + opacity_range * progress
+	return 1.0 - opacity_range * progress
+
+
+def preloader_status_text(
+	value: str,
+	step: int | None = None,
+	total: int | None = None,
+) -> str:
+	"""Render one launch stage without letting its widget change shape."""
+
+	if step is None or total is None:
+		return value
+	return f"Step {step} of {total} · {value}"
+
+
+def preloader_logo_subsample_factor(
+	width: int,
+	height: int,
+	*,
+	max_dimension: int = 88,
+) -> int:
+	"""Return an integer Tk subsample factor for a compact preloader logo."""
+
+	if width <= 0 or height <= 0 or max_dimension <= 0:
+		raise ValueError("preloader logo dimensions must be positive")
+	return max(
+		1,
+		math.ceil(width / max_dimension),
+		math.ceil(height / max_dimension),
+	)
+
+
+def startup_progress_animation_value(current: float, target: float) -> float:
+	"""Ease one determinate startup-progress frame without overshooting."""
+
+	distance = abs(target - current)
+	if distance == 0:
+		return target
+	step = max(
+		STARTUP_PROGRESS_ANIMATION_MINIMUM_STEP,
+		distance * STARTUP_PROGRESS_ANIMATION_EASING,
+	)
+	if step >= distance:
+		return target
+	direction = 1.0 if target > current else -1.0
+	return current + (direction * step)
+
+
+def freeze_preloader_geometry(frame: Frame) -> tuple[int, int]:
+	"""Lock a measured preloader frame so later status text cannot reflow it."""
+
+	frame.update_idletasks()
+	width = max(1, frame.winfo_reqwidth())
+	height = max(1, frame.winfo_reqheight())
+	frame.configure(width=width, height=height)
+	frame.pack_propagate(False)
+	return width, height
+
+
+def launcher_restart_spec(
+	paths: AppPaths,
+	environment: Mapping[str, str] | None = None,
+) -> tuple[list[str], dict[str, str]]:
+	"""Build an isolated restart that imports the newly activated release."""
+
+	current = paths.launcher_current
+	if not current.is_symlink():
+		raise DashboardError(f"launcher current path is not a symbolic link: {current}")
+	try:
+		release = current.resolve(strict=True)
+	except OSError as exc:
+		raise DashboardError(f"launcher current release is unavailable: {exc}") from exc
+	releases = paths.launcher_releases.resolve(strict=False)
+	if release.parent != releases or not (release / "ptlinux").is_dir():
+		raise DashboardError(f"launcher current release is outside the release store: {release}")
+	env = dict(os.environ if environment is None else environment)
+	env.update(
+		{
+			"PTLINUX_INSTALL_DIR": str(paths.data),
+			"PTLINUX_APP_HOME": str(paths.app_home),
+			"PYTHONSAFEPATH": "1",
+			"PYTHONPATH": str(release),
+		}
+	)
+	return [sys.executable, "-B", "-P", "-m", "ptlinux", "gui"], env
+
+
+def launcher_child_command(*arguments: str) -> list[str]:
+	"""Build a child command without writing bytecode into the active release."""
+
+	return [sys.executable, "-B", "-m", "ptlinux", *arguments]
+
+
+def dashboard_poll_interval(
+	*,
+	operation_active: bool,
+	game_window_active: bool,
+	pending_events: bool,
+) -> int:
+	"""Choose a responsive poll rate while avoiding busy idle/game polling."""
+
+	if pending_events:
+		return POLL_INTERVAL_MILLISECONDS
+	if operation_active and not game_window_active:
+		return ACTIVE_POLL_INTERVAL_MILLISECONDS
+	return IDLE_POLL_INTERVAL_MILLISECONDS
--- a/ptlinux/ui/viewmodels/downloader.py
+++ b/ptlinux/ui/viewmodels/downloader.py
@@ -0,0 +1,240 @@
+"""Download and long-running-operation presentation state."""
+
+from __future__ import annotations
+
+from collections import deque
+import math
+import threading
+import time
+from typing import Mapping
+
+
+DOWNLOAD_SPEED_PRESETS_MBPS: dict[str, float | None] = {
+	"Unlimited": None,
+	"5 MB/s": 5.0,
+	"10 MB/s": 10.0,
+	"25 MB/s": 25.0,
+	"50 MB/s": 50.0,
+}
+DOWNLOAD_SPEED_CUSTOM_LABEL = "Custom…"
+TRANSFER_RATE_WINDOW_SECONDS = 3.0
+TRANSFER_RATE_MINIMUM_INTERVAL_SECONDS = 0.2
+TRANSFER_RATE_IDLE_SECONDS = 2.0
+
+
+def download_speed_setting_label(limit_mbps: float | None) -> str:
+	for label, preset in DOWNLOAD_SPEED_PRESETS_MBPS.items():
+		if preset == limit_mbps:
+			return label
+	return DOWNLOAD_SPEED_CUSTOM_LABEL
+
+
+def download_speed_setting_value(label: str, custom_value: str) -> float | None:
+	if label in DOWNLOAD_SPEED_PRESETS_MBPS:
+		return DOWNLOAD_SPEED_PRESETS_MBPS[label]
+	if label != DOWNLOAD_SPEED_CUSTOM_LABEL:
+		raise ValueError(f"unsupported download speed choice: {label}")
+	try:
+		value = float(custom_value.strip())
+	except ValueError as error:
+		raise ValueError("custom download speed must be a number in MB/s") from error
+	if not math.isfinite(value) or value <= 0 or value > 10_000:
+		raise ValueError(
+			"custom download speed must be greater than 0 and at most 10000 MB/s"
+		)
+	return value
+
+
+def download_activity_metadata(
+	*,
+	current_server: str | None,
+	retry_count: int,
+	retry_reason: str | None,
+	active_workers: int,
+	target_workers: int,
+) -> str:
+	"""Render compact download details for the existing activity heading."""
+
+	if retry_count < 0 or active_workers < 0 or target_workers < 0:
+		raise ValueError("download activity counts must not be negative")
+	parts: list[str] = []
+	if current_server:
+		parts.append(f"CDN: {_ellipsize(current_server, 28)}")
+	if target_workers:
+		parts.append(f"Workers: {active_workers}/{target_workers}")
+	retry_text = f"Retries: {retry_count}"
+	if retry_reason:
+		retry_text += f" ({_ellipsize(retry_reason, 26, keep_tail=True)})"
+	parts.append(retry_text)
+	return " · ".join(parts)
+
+
+def _ellipsize(value: str, limit: int, *, keep_tail: bool = False) -> str:
+	normalized = " ".join(value.split())
+	if len(normalized) <= limit:
+		return normalized
+	if keep_tail:
+		return f"…{normalized[-(limit - 1):]}"
+	return f"{normalized[:limit - 1]}…"
+
+
+def operation_activity_text(
+	message: str,
+	*,
+	started_at: float | None,
+	step: int = 0,
+	total: int = 0,
+	detail: str | None = None,
+	timestamp: float | None = None,
+) -> str:
+	"""Render live operation progress without dropping its elapsed timer."""
+
+	if detail is not None:
+		base = detail
+	elif total > 0:
+		base = f"Step {step}/{total} · {message}"
+	else:
+		return message
+	if started_at is None:
+		return base
+	now = time.monotonic() if timestamp is None else timestamp
+	elapsed = int(max(0.0, now - started_at))
+	return f"{base} · {elapsed}s elapsed"
+
+
+def should_refresh_operation_elapsed(
+	*,
+	active: bool,
+	started_at: float | None,
+	stage_total: int,
+	detail: str | None,
+) -> bool:
+	"""Return whether the dashboard has operation text with a live timer."""
+
+	return (
+		active
+		and started_at is not None
+		and (stage_total > 0 or detail is not None)
+	)
+
+
+class ProgressCoalescer:
+	"""Keep only the latest cross-thread progress event for the Tk loop."""
+
+	def __init__(self) -> None:
+		self._lock = threading.Lock()
+		self._latest: dict[str, object] | None = None
+
+	def publish(self, event: Mapping[str, object]) -> None:
+		with self._lock:
+			self._latest = dict(event)
+
+	def take(self) -> dict[str, object] | None:
+		with self._lock:
+			latest = self._latest
+			self._latest = None
+			return latest
+
+	def clear(self) -> None:
+		with self._lock:
+			self._latest = None
+
+
+class TransferRateEstimator:
+	"""Estimate a stable byte rate from cumulative network-transfer samples."""
+
+	def __init__(
+		self,
+		window_seconds: float = TRANSFER_RATE_WINDOW_SECONDS,
+		minimum_interval_seconds: float = TRANSFER_RATE_MINIMUM_INTERVAL_SECONDS,
+	) -> None:
+		if window_seconds <= 0:
+			raise ValueError("transfer-rate window must be positive")
+		if minimum_interval_seconds < 0:
+			raise ValueError("transfer-rate minimum interval must not be negative")
+		self._window_seconds = window_seconds
+		self._minimum_interval_seconds = minimum_interval_seconds
+		self._samples: deque[tuple[float, int]] = deque()
+
+	def reset(self) -> None:
+		self._samples.clear()
+
+	def update(
+		self,
+		transferred_bytes: int,
+		*,
+		timestamp: float | None = None,
+	) -> float | None:
+		"""Record one cumulative sample and return bytes per second when stable."""
+
+		if transferred_bytes < 0:
+			raise ValueError("transferred bytes must not be negative")
+		now = time.monotonic() if timestamp is None else timestamp
+		if self._samples:
+			last_time, last_bytes = self._samples[-1]
+			if now <= last_time or transferred_bytes < last_bytes:
+				self.reset()
+		self._samples.append((now, transferred_bytes))
+		cutoff = now - self._window_seconds
+		while len(self._samples) > 2 and self._samples[1][0] <= cutoff:
+			self._samples.popleft()
+		if len(self._samples) < 2:
+			return None
+		first_time, first_bytes = self._samples[0]
+		elapsed = now - first_time
+		if elapsed < self._minimum_interval_seconds:
+			return None
+		return max(0.0, (transferred_bytes - first_bytes) / elapsed)
+
+	def is_idle(
+		self,
+		threshold_seconds: float = TRANSFER_RATE_IDLE_SECONDS,
+		*,
+		timestamp: float | None = None,
+	) -> bool:
+		"""Return whether no cumulative sample arrived within the threshold."""
+
+		if threshold_seconds < 0:
+			raise ValueError("idle threshold must not be negative")
+		if not self._samples:
+			return False
+		now = time.monotonic() if timestamp is None else timestamp
+		return now - self._samples[-1][0] >= threshold_seconds
+
+
+def as_non_negative_int(value: object) -> int:
+	return value if isinstance(value, int) and value >= 0 else 0
+
+
+def as_non_negative_float(value: object) -> float | None:
+	if isinstance(value, bool) or not isinstance(value, (int, float)):
+		return None
+	number = float(value)
+	return number if math.isfinite(number) and number >= 0 else None
+
+
+def format_bytes(value: int) -> str:
+	units = ("B", "KiB", "MiB", "GiB")
+	amount = float(value)
+	for unit in units:
+		if amount < 1024 or unit == units[-1]:
+			return f"{amount:.1f} {unit}" if unit != "B" else f"{int(amount)} {unit}"
+		amount /= 1024
+	return f"{value} B"
+
+
+def format_transfer_rate(bytes_per_second: float) -> str:
+	return f"{format_bytes(max(0, round(bytes_per_second)))}/s"
+
+
+def format_eta(seconds: float | None) -> str:
+	if seconds is None or not math.isfinite(seconds) or seconds < 0:
+		return "calculating…"
+	total_seconds = max(0, math.ceil(seconds))
+	hours, remainder = divmod(total_seconds, 60 * 60)
+	minutes, seconds = divmod(remainder, 60)
+	if hours:
+		return f"{hours}h {minutes:02d}m"
+	if minutes:
+		return f"{minutes}m {seconds:02d}s"
+	return f"{seconds}s"
--- a/ptlinux/ui/viewmodels/health.py
+++ b/ptlinux/ui/viewmodels/health.py
@@ -0,0 +1,122 @@
+"""Official server-health presentation and request state."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import time
+from typing import Callable
+
+from ...server_status import RealmServerStatus, ServerStatusError, fetch_server_status
+
+
+SERVER_STATUS_STALE_SECONDS = 5 * 60
+
+
+@dataclass(frozen=True)
+class ServerStatusPresentation:
+	"""Text and Tk style for one official realm-health result."""
+
+	text: str
+	style: str
+
+
+@dataclass(frozen=True)
+class ServerStatusCheck:
+	"""One measured attempt to retrieve official realm health."""
+
+	statuses: tuple[RealmServerStatus, ...] | None
+	checked_at: float
+	duration_seconds: float
+	error: str | None = None
+
+
+def perform_server_status_check(
+	*,
+	fetcher: Callable[[], tuple[RealmServerStatus, ...]] = fetch_server_status,
+	monotonic: Callable[[], float] = time.monotonic,
+	wall_clock: Callable[[], float] = time.time,
+) -> ServerStatusCheck:
+	"""Measure one official status request without leaking its endpoint."""
+
+	started_at = monotonic()
+	try:
+		statuses = fetcher()
+	except ServerStatusError as error:
+		duration_seconds = max(0.0, monotonic() - started_at)
+		return ServerStatusCheck(
+			statuses=None,
+			checked_at=wall_clock(),
+			duration_seconds=duration_seconds,
+			error=str(error),
+		)
+	duration_seconds = max(0.0, monotonic() - started_at)
+	return ServerStatusCheck(
+		statuses=statuses,
+		checked_at=wall_clock(),
+		duration_seconds=duration_seconds,
+	)
+
+
+def server_status_check_text(
+	check: ServerStatusCheck,
+	*,
+	timestamp: float | None = None,
+) -> str:
+	"""Render fetch age, latency, and freshness for the hidden debug view."""
+
+	now = time.time() if timestamp is None else timestamp
+	age_seconds = max(0.0, now - check.checked_at)
+	whole_seconds = int(age_seconds)
+	if whole_seconds < 60:
+		age = f"{whole_seconds}s"
+	elif whole_seconds < 60 * 60:
+		age = f"{whole_seconds // 60}m"
+	elif whole_seconds < 24 * 60 * 60:
+		age = f"{whole_seconds // (60 * 60)}h"
+	else:
+		age = f"{whole_seconds // (24 * 60 * 60)}d"
+	latency_milliseconds = max(0, round(check.duration_seconds * 1000))
+	timing = f"checked {age} ago · {latency_milliseconds} ms"
+	if check.error is not None:
+		return f"failed · {timing} · {check.error}"
+	freshness = "stale" if age_seconds >= SERVER_STATUS_STALE_SECONDS else "fresh"
+	return f"{freshness} · {timing}"
+
+
+def _server_component_text(name: str, online: bool | None) -> str:
+	if online is True:
+		state = "up"
+	elif online is False:
+		state = "down"
+	else:
+		state = "unknown"
+	return f"{name} {state}"
+
+
+def server_status_presentation(
+	status: RealmServerStatus,
+	*,
+	show_debug_details: bool = False,
+) -> ServerStatusPresentation:
+	if status.online is True:
+		text = f"● {status.name}"
+		style = "ServerOnline.TLabel"
+	elif status.online is False:
+		text = f"● {status.name} · offline"
+		style = "ServerOffline.TLabel"
+	else:
+		text = f"○ {status.name} · status unknown"
+		style = "ServerUnknown.TLabel"
+	if show_debug_details:
+		details: list[str] = []
+		if status.players_online is not None:
+			label = "player" if status.players_online == 1 else "players"
+			details.append(f"{status.players_online} {label}")
+		details.extend(
+			(
+				_server_component_text("login", status.login_online),
+				_server_component_text("game", status.game_online),
+			)
+		)
+		text = " · ".join((text, *details))
+	return ServerStatusPresentation(text=text, style=style)
--- a/ptlinux/ui/viewmodels/installation.py
+++ b/ptlinux/ui/viewmodels/installation.py
@@ -0,0 +1,46 @@
+"""Presentation state for managed-game installation operations."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class GameSyncPresentation:
+	"""User-facing wording for one managed-game synchronization."""
+
+	operation_label: str
+	file_progress_verb: str
+
+
+def game_sync_presentation(
+	*,
+	installed: bool,
+	repair: bool = False,
+) -> GameSyncPresentation:
+	"""Distinguish a fresh download from an update or repair."""
+
+	if not installed:
+		return GameSyncPresentation(
+			operation_label=(
+				"Installing and verifying official game files…"
+				if repair
+				else "Installing official game files…"
+			),
+			file_progress_verb="Downloading",
+		)
+	if repair:
+		return GameSyncPresentation(
+			operation_label="Checking every game file against the official depot…",
+			file_progress_verb="Repairing",
+		)
+	return GameSyncPresentation(
+		operation_label="Synchronizing official game files…",
+		file_progress_verb="Updating",
+	)
+
+
+def game_file_progress_status(current_file: str, verb: str) -> str:
+	"""Render the current depot asset with the operation's exact intent."""
+
+	return f"{verb} {current_file}"
				`@@ -0,0 +1 @@`
				`"""Pure state and presentation models used by the native dashboard."""`