pristontale-eu-linux/ptlinux/self_update.py

"""Signed, atomic self-updates for the native Linux launcher."""

from __future__ import annotations

from contextlib import contextmanager
from dataclasses import asdict, dataclass, replace
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import shutil
import stat
import subprocess
import sys
import tarfile
import tempfile
import time
from typing import Iterator, Mapping
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit, urlunsplit
from urllib.request import Request, urlopen

from . import __version__
from .locking import launcher_operation_lock, managed_game_process_is_active
from .paths import AppPaths
from .telemetry import (
	classify_network_failure,
	current_telemetry,
	network_http_status,
)
from .update_resolver import (
	ACTIVE_STATUS,
	PRERELEASE_CHANNEL,
	STABLE_CHANNEL,
	SUPPORTED_RELEASE_STATUSES,
	ReleaseModelError,
	ReleaseRecord,
	SemanticVersion,
	UpdateResolver,
	validate_channel,
	validate_channel_version,
)


DEFAULT_UPDATE_REPOSITORY_URL = (
	"https://git.m4ko.net/m4ko/pristontale-eu-linux"
)
DEFAULT_STABLE_UPDATE_MANIFEST_URL = (
	f"{DEFAULT_UPDATE_REPOSITORY_URL}/raw/branch/main/updates/stable.json"
)
DEFAULT_PRERELEASE_UPDATE_MANIFEST_URL = (
	f"{DEFAULT_UPDATE_REPOSITORY_URL}/raw/branch/main/updates/prerelease.json"
)
DEFAULT_UPDATE_HISTORY_URL = (
	f"{DEFAULT_UPDATE_REPOSITORY_URL}/raw/branch/main/updates/history.json"
)
# Retain the original name for callers importing the stable default.
DEFAULT_UPDATE_MANIFEST_URL = DEFAULT_STABLE_UPDATE_MANIFEST_URL
UPDATE_MANIFEST_URL_VARIABLE = "PTLINUX_UPDATE_MANIFEST_URL"
UPDATE_STABLE_MANIFEST_URL_VARIABLE = "PTLINUX_UPDATE_STABLE_MANIFEST_URL"
UPDATE_PRERELEASE_MANIFEST_URL_VARIABLE = "PTLINUX_UPDATE_PRERELEASE_MANIFEST_URL"
UPDATE_HISTORY_URL_VARIABLE = "PTLINUX_UPDATE_HISTORY_URL"
UPDATE_PUBLIC_KEY_VARIABLE = "PTLINUX_UPDATE_PUBLIC_KEY"
PUBLIC_KEY_FILENAME = "launcher-update-public.pem"
UPDATE_APPLICATION = "PristonTale-EU"
USER_AGENT = f"PT-Linux/{__version__} launcher-updater"
READ_SIZE = 256 * 1024
MAX_MANIFEST_BYTES = 128 * 1024
MAX_HISTORY_BYTES = 512 * 1024
MAX_SIGNATURE_BYTES = 16 * 1024
MAX_EMBEDDED_RELEASE_NOTES_BYTES = 64 * 1024
MAX_ARCHIVE_BYTES = 32 * 1024 * 1024
MAX_EXTRACTED_BYTES = 64 * 1024 * 1024
MAX_ARCHIVE_MEMBERS = 512
UPDATE_ARCHIVE_PREFIX = "pristontale-eu-linux-update-v"
_SHA256_PATTERN = re.compile(r"[0-9a-f]{64}\Z")
_ALLOWED_RELEASE_ROOTS = {"ptlinux", "ptdiag", "release.json"}
UPDATE_CHECK_CACHE_SCHEMA = 2
RELEASE_HISTORY_SCHEMA = 1
MAX_RELEASE_HISTORY = 200
LOCAL_OPERATION_HISTORY_SCHEMA = 1
MAX_LOCAL_OPERATION_HISTORY = 30


class LauncherUpdateError(RuntimeError):
	"""Raised when a launcher update cannot be trusted or installed safely."""


@dataclass(frozen=True)
class ManifestFile:
	path: str
	size: int
	sha256: str


@dataclass(frozen=True)
class UpdateManifest:
	schema: int
	channel: str
	version: str
	minimum_updater_version: str
	archive_url: str
	archive_sha256: str
	archive_size: int
	published_at: str
	notes_url: str | None
	tag: str | None = None
	commit: str | None = None
	status: str = ACTIVE_STATUS
	title: str | None = None
	summary: str | None = None
	notes_markdown: str | None = None
	archive_name: str | None = None
	files: tuple[ManifestFile, ...] = ()
	manifest_sha256: str | None = None
	config_schema: int = 1
	minimum_config_schema: int = 1


@dataclass(frozen=True)
class LauncherUpdateCheck:
	current_version: str
	latest_version: str
	available: bool
	supported: bool
	reason: str
	manifest_url: str
	manifest: UpdateManifest
	checked_at: str | None = None
	selected_channel: str = STABLE_CHANNEL


@dataclass(frozen=True)
class LauncherUpdateResult:
	changed: bool
	current_version: str
	installed_version: str
	release_directory: str | None
	previous_release: str | None
	archive: str | None
	archive_sha256: str | None
	restart_required: bool
	state_recorded: bool
	warning: str | None
	selected_channel: str = STABLE_CHANNEL
	installed_channel: str = STABLE_CHANNEL
	tag: str | None = None
	manifest_schema: int | None = None
	manifest_sha256: str | None = None


@dataclass(frozen=True)
class LauncherUpdateVerification:
	current_version: str
	release_version: str
	manifest_url: str
	archive_url: str
	archive_size: int
	archive_sha256: str
	payload_files: int
	channel: str = STABLE_CHANNEL
	tag: str | None = None
	manifest_sha256: str | None = None


@dataclass(frozen=True)
class LauncherRollbackResult:
	from_version: str
	to_version: str
	current_release: str
	previous_release: str
	restart_required: bool
	from_channel: str = STABLE_CHANNEL
	to_channel: str = STABLE_CHANNEL
	state_recorded: bool = True
	warning: str | None = None


@dataclass(frozen=True)
class LauncherInstalledRelease:
	"""One validated release retained in the launcher's version store."""

	version: str
	installed_at: str
	directory: str
	active: bool
	previous: bool
	channel: str = STABLE_CHANNEL


@dataclass(frozen=True)
class LauncherUpdateHistory:
	"""Validated launcher releases available for inspection or rollback."""

	current_version: str | None
	previous_version: str | None
	releases: tuple[LauncherInstalledRelease, ...]


@dataclass(frozen=True)
class LauncherRemoteRelease:
	"""One independently installable release from signed remote history."""

	version: str
	channel: str
	tag: str
	manifest_url: str
	manifest_signature_url: str
	release_date: str
	release_notes_url: str
	status: str
	title: str
	release_id: int | None = None
	config_schema: int | None = None
	minimum_config_schema: int | None = None

	def resolver_record(self) -> ReleaseRecord:
		return ReleaseRecord(
			version=self.version,
			channel=self.channel,
			status=self.status,
			tag=self.tag,
			manifest_url=self.manifest_url,
			manifest_signature_url=self.manifest_signature_url,
			release_date=self.release_date,
			release_notes_url=self.release_notes_url,
			compatibility={
				"config_schema": self.config_schema,
				"minimum_config_schema": self.minimum_config_schema,
			},
		)


@dataclass(frozen=True)
class LauncherRemoteHistory:
	"""Authenticated release catalog used for history and remote rollback."""

	generated_at: str
	history_url: str
	releases: tuple[LauncherRemoteRelease, ...]
	history_sha256: str


def default_public_key_path() -> Path:
	return Path(__file__).with_name(PUBLIC_KEY_FILENAME)


def configured_manifest_url(
	environment: Mapping[str, str] | None = None,
	*,
	channel: str = STABLE_CHANNEL,
) -> str:
	"""Return the direct channel URL retained for legacy callers."""

	urls = configured_manifest_urls(channel=channel, environment=environment)
	return urls[-1][1]


def configured_manifest_urls(
	*,
	channel: str = STABLE_CHANNEL,
	environment: Mapping[str, str] | None = None,
) -> tuple[tuple[str, str, bool], ...]:
	"""Return channel, URL, and optionality for authenticated discovery."""

	try:
		validate_channel(channel)
	except ReleaseModelError as exc:
		raise LauncherUpdateError(str(exc)) from exc
	env = os.environ if environment is None else environment
	generic_override = env.get(UPDATE_MANIFEST_URL_VARIABLE)
	if generic_override:
		return ((channel, generic_override, False),)
	stable_url = env.get(
		UPDATE_STABLE_MANIFEST_URL_VARIABLE,
		DEFAULT_STABLE_UPDATE_MANIFEST_URL,
	)
	if channel == STABLE_CHANNEL:
		return ((STABLE_CHANNEL, stable_url, False),)
	prerelease_url = env.get(
		UPDATE_PRERELEASE_MANIFEST_URL_VARIABLE,
		DEFAULT_PRERELEASE_UPDATE_MANIFEST_URL,
	)
	return (
		(STABLE_CHANNEL, stable_url, False),
		(PRERELEASE_CHANNEL, prerelease_url, True),
	)


def configured_public_key_path(
	environment: Mapping[str, str] | None = None,
) -> Path:
	env = os.environ if environment is None else environment
	configured = env.get(UPDATE_PUBLIC_KEY_VARIABLE)
	return Path(configured).expanduser() if configured else default_public_key_path()


def configured_history_url(
	environment: Mapping[str, str] | None = None,
) -> str:
	env = os.environ if environment is None else environment
	return env.get(UPDATE_HISTORY_URL_VARIABLE, DEFAULT_UPDATE_HISTORY_URL)


def fetch_launcher_release_history(
	*,
	history_url: str | None = None,
	public_key_path: Path | None = None,
	environment: Mapping[str, str] | None = None,
	timeout_seconds: float = 20.0,
) -> LauncherRemoteHistory:
	"""Fetch and authenticate the immutable-release catalog used for rollback."""

	url = history_url or configured_history_url(environment)
	_require_https_url(url, "release history URL")
	key = public_key_path or configured_public_key_path(environment)
	history_bytes = _fetch_small_file(
		url,
		maximum_bytes=MAX_HISTORY_BYTES,
		timeout_seconds=timeout_seconds,
		label="release history",
	)
	signature_bytes = _fetch_small_file(
		_signature_url(url),
		maximum_bytes=MAX_SIGNATURE_BYTES,
		timeout_seconds=timeout_seconds,
		label="release history signature",
	)
	_verify_manifest_signature(history_bytes, signature_bytes, key)
	return parse_launcher_release_history(history_bytes, history_url=url)


def cache_launcher_release_history(
	cache_path: Path,
	history: LauncherRemoteHistory,
	*,
	public_key_path: Path | None = None,
	environment: Mapping[str, str] | None = None,
) -> bool:
	"""Cache authenticated history for display; installation always re-fetches it."""

	key = public_key_path or configured_public_key_path(environment)
	try:
		_write_json(
			cache_path,
			{
				"schema": RELEASE_HISTORY_SCHEMA,
				"history_url": history.history_url,
				"history_sha256": history.history_sha256,
				"public_key_sha256": _sha256_file(key.expanduser().resolve()),
				"generated_at": history.generated_at,
				"releases": [
					_remote_release_to_history_dict(release)
					for release in history.releases
				],
			},
		)
	except (OSError, RuntimeError):
		return False
	return True


def read_cached_launcher_release_history(
	cache_path: Path,
	*,
	history_url: str | None = None,
	public_key_path: Path | None = None,
	environment: Mapping[str, str] | None = None,
) -> LauncherRemoteHistory | None:
	"""Read display-only history previously authenticated with the current key."""

	url = history_url or configured_history_url(environment)
	key = public_key_path or configured_public_key_path(environment)
	try:
		_require_https_url(url, "release history URL")
		value = json.loads(cache_path.read_text(encoding="utf-8"))
		key_sha256 = _sha256_file(key.expanduser().resolve())
	except (
		OSError,
		UnicodeError,
		json.JSONDecodeError,
		LauncherUpdateError,
		RuntimeError,
	):
		return None
	if not isinstance(value, dict):
		return None
	if value.get("schema") != RELEASE_HISTORY_SCHEMA:
		return None
	if value.get("history_url") != url or value.get("public_key_sha256") != key_sha256:
		return None
	stored_digest = value.get("history_sha256")
	if not isinstance(stored_digest, str) or _SHA256_PATTERN.fullmatch(stored_digest) is None:
		return None
	try:
		serialized = {
			"application": UPDATE_APPLICATION,
			"generated_at": value.get("generated_at"),
			"releases": value.get("releases"),
			"schema": RELEASE_HISTORY_SCHEMA,
		}
		history = parse_launcher_release_history(
			(json.dumps(serialized, sort_keys=True) + "\n").encode("utf-8"),
			history_url=url,
		)
	except (LauncherUpdateError, TypeError, ValueError):
		return None
	return replace(history, history_sha256=stored_digest)


def _remote_release_to_history_dict(
	release: LauncherRemoteRelease,
) -> dict[str, object]:
	return {
		"channel": release.channel,
		"compatibility": {
			"config_schema": release.config_schema,
			"minimum_config_schema": release.minimum_config_schema,
		},
		"manifest_signature_url": release.manifest_signature_url,
		"manifest_url": release.manifest_url,
		"release_date": release.release_date,
		"release_id": release.release_id,
		"release_notes_url": release.release_notes_url,
		"status": release.status,
		"tag": release.tag,
		"title": release.title,
		"version": release.version,
	}


def check_for_launcher_update(
	*,
	current_version: str = __version__,
	manifest_url: str | None = None,
	public_key_path: Path | None = None,
	environment: Mapping[str, str] | None = None,
	timeout_seconds: float = 20.0,
	channel: str = STABLE_CHANNEL,
) -> LauncherUpdateCheck:
	"""Fetch authenticated channel metadata and resolve the greatest candidate."""

	_parse_version(current_version, "current launcher version")
	try:
		validate_channel(channel)
	except ReleaseModelError as exc:
		raise LauncherUpdateError(str(exc)) from exc
	configured = (
		((channel, manifest_url, False),)
		if manifest_url is not None
		else configured_manifest_urls(channel=channel, environment=environment)
	)
	key = public_key_path or configured_public_key_path(environment)
	candidates: list[tuple[UpdateManifest, str]] = []
	for expected_channel, url, optional in configured:
		_require_https_url(url, "update manifest URL")
		try:
			manifest = _fetch_authenticated_manifest(
				url,
				key,
				timeout_seconds=timeout_seconds,
			)
		except LauncherUpdateError as exc:
			if optional and "HTTP 404" in str(exc):
				continue
			raise
		if manifest.channel != expected_channel:
			raise LauncherUpdateError(
				f"{expected_channel} channel URL returned a {manifest.channel} manifest"
			)
		candidates.append((manifest, url))
	if not candidates:
		raise LauncherUpdateError(f"no authenticated {channel} release metadata is available")
	try:
		resolver = UpdateResolver(
			ReleaseRecord(
				version=manifest.version,
				channel=manifest.channel,
				status=manifest.status,
				tag=manifest.tag,
				manifest_url=url,
			)
			for manifest, url in candidates
		)
		selected = resolver.latest_eligible(channel)
	except ReleaseModelError as exc:
		raise LauncherUpdateError(str(exc)) from exc
	if selected is None:
		# Preserve authenticated metadata for a clear withdrawn/blocked status.
		manifest, url = max(
			candidates,
			key=lambda item: _parse_version(item[0].version, "manifest version"),
		)
	else:
		manifest, url = next(
			item for item in candidates if item[0].version == selected.version
		)
	check = _evaluate_launcher_update(
		manifest,
		current_version,
		url,
		checked_at=datetime.now().astimezone().isoformat(timespec="seconds"),
		selected_channel=channel,
	)
	_send_launcher_release_event(
		"update_release_discovered",
		metadata=_launcher_release_metadata(
			current_version=current_version,
			target_version=manifest.version,
			current_channel=SemanticVersion.parse(current_version).channel,
			target_channel=manifest.channel,
			selected_channel=channel,
			tag=manifest.tag,
			manifest_schema=manifest.schema,
			manifest_sha256=manifest.manifest_sha256,
			rollback=False,
		),
		operation="launcher_update",
		stage="discovery",
	)
	return check


def _fetch_authenticated_manifest(
	url: str,
	public_key_path: Path,
	*,
	timeout_seconds: float,
) -> UpdateManifest:
	manifest_bytes = _fetch_small_file(
		url,
		maximum_bytes=MAX_MANIFEST_BYTES,
		timeout_seconds=timeout_seconds,
		label="update manifest",
	)
	signature_bytes = _fetch_small_file(
		_signature_url(url),
		maximum_bytes=MAX_SIGNATURE_BYTES,
		timeout_seconds=timeout_seconds,
		label="update manifest signature",
	)
	_verify_manifest_signature(manifest_bytes, signature_bytes, public_key_path)
	return parse_update_manifest(manifest_bytes)


def _fetch_historical_manifest(
	release: LauncherRemoteRelease,
	public_key_path: Path,
	*,
	timeout_seconds: float,
) -> UpdateManifest:
	manifest_bytes = _fetch_small_file(
		release.manifest_url,
		maximum_bytes=MAX_MANIFEST_BYTES,
		timeout_seconds=timeout_seconds,
		label=f"launcher {release.version} manifest",
	)
	signature_bytes = _fetch_small_file(
		release.manifest_signature_url,
		maximum_bytes=MAX_SIGNATURE_BYTES,
		timeout_seconds=timeout_seconds,
		label=f"launcher {release.version} manifest signature",
	)
	_verify_manifest_signature(manifest_bytes, signature_bytes, public_key_path)
	manifest = parse_update_manifest(manifest_bytes)
	for field, actual, expected in (
		("version", manifest.version, release.version),
		("channel", manifest.channel, release.channel),
		("tag", manifest.tag, release.tag),
	):
		if actual != expected:
				raise LauncherUpdateError(
					f"historical manifest {field} {actual!r} does not match "
					f"signed history {expected!r}"
				)
	if (
		release.config_schema is not None
		or release.minimum_config_schema is not None
	):
		if release.config_schema is None or release.minimum_config_schema is None:
			raise LauncherUpdateError(
				"signed history has incomplete config compatibility metadata"
			)
		manifest = replace(
			manifest,
			config_schema=release.config_schema,
			minimum_config_schema=release.minimum_config_schema,
		)
	return manifest


def read_cached_launcher_update_check(
	cache_path: Path,
	*,
	current_version: str = __version__,
	manifest_url: str | None = None,
	public_key_path: Path | None = None,
	environment: Mapping[str, str] | None = None,
	channel: str = STABLE_CHANNEL,
) -> LauncherUpdateCheck | None:
	"""Read a display-only result from the last authenticated live check."""

	try:
		validate_channel(channel)
	except ReleaseModelError:
		return None
	allowed_urls = (
		{manifest_url}
		if manifest_url is not None
		else {url for _source_channel, url, _optional in configured_manifest_urls(
			channel=channel,
			environment=environment,
		)}
	)
	key = public_key_path or configured_public_key_path(environment)
	try:
		_parse_version(current_version, "current launcher version")
		for allowed_url in allowed_urls:
			_require_https_url(allowed_url, "update manifest URL")
		key_sha256 = _sha256_file(key.expanduser().resolve())
		value = json.loads(cache_path.read_text(encoding="utf-8"))
	except (
		OSError,
		UnicodeError,
		json.JSONDecodeError,
		LauncherUpdateError,
		RuntimeError,
	):
		return None
	if not isinstance(value, dict):
		return None
	if value.get("schema") != UPDATE_CHECK_CACHE_SCHEMA:
		return None
	if value.get("current_version") != current_version:
		return None
	url = value.get("manifest_url")
	if not isinstance(url, str) or url not in allowed_urls:
		return None
	if value.get("selected_channel") != channel:
		return None
	if value.get("public_key_sha256") != key_sha256:
		return None
	manifest_value = value.get("manifest")
	if not isinstance(manifest_value, dict):
		return None
	checked_at = _cached_timestamp(value.get("checked_at"))
	if checked_at is None:
		return None
	try:
		manifest = parse_update_manifest(
			(json.dumps(manifest_value, sort_keys=True) + "\n").encode("utf-8")
		)
		cached_manifest_sha256 = manifest_value.get("manifest_sha256")
		if (
			isinstance(cached_manifest_sha256, str)
			and _SHA256_PATTERN.fullmatch(cached_manifest_sha256) is not None
		):
			manifest = replace(
				manifest,
				manifest_sha256=cached_manifest_sha256,
			)
	except (LauncherUpdateError, TypeError, ValueError):
		return None
	return _evaluate_launcher_update(
		manifest,
		current_version,
		url,
		checked_at=checked_at,
		selected_channel=channel,
	)


def cache_launcher_update_check(
	cache_path: Path,
	check: LauncherUpdateCheck,
	*,
	public_key_path: Path | None = None,
	environment: Mapping[str, str] | None = None,
) -> bool:
	"""Cache an authenticated result for startup display, never for installation."""

	key = public_key_path or configured_public_key_path(environment)
	try:
		key_sha256 = _sha256_file(key.expanduser().resolve())
		_write_json(
			cache_path,
			{
				"schema": UPDATE_CHECK_CACHE_SCHEMA,
				"checked_at": check.checked_at
				or datetime.now().astimezone().isoformat(timespec="seconds"),
				"current_version": check.current_version,
				"manifest_url": check.manifest_url,
				"selected_channel": check.selected_channel,
				"public_key_sha256": key_sha256,
				"manifest": asdict(check.manifest),
			},
		)
	except (OSError, RuntimeError):
		return False
	return True


def _evaluate_launcher_update(
	manifest: UpdateManifest,
	current_version: str,
	manifest_url: str,
	*,
	checked_at: str | None = None,
	selected_channel: str = STABLE_CHANNEL,
) -> LauncherUpdateCheck:
	try:
		validate_channel(selected_channel)
	except ReleaseModelError as exc:
		raise LauncherUpdateError(str(exc)) from exc
	current = _parse_version(current_version, "current launcher version")
	latest = _parse_version(manifest.version, "manifest version")
	minimum = _parse_version(
		manifest.minimum_updater_version,
		"minimum updater version",
	)
	available = latest > current and manifest.status == ACTIVE_STATUS
	supported = current >= minimum
	if manifest.status != ACTIVE_STATUS:
		reason = f"Launcher {manifest.version} is {manifest.status} and will not be offered."
	elif not available:
		reason = (
			"The selected channel will not downgrade this launcher."
			if latest < current
			else "The launcher is up to date."
		)
	elif not supported:
		reason = (
			f"Version {manifest.version} requires updater "
			f"{manifest.minimum_updater_version} or newer; install it manually."
		)
	else:
		reason = f"Launcher {manifest.version} is available."
	return LauncherUpdateCheck(
		current_version=current_version,
		latest_version=manifest.version,
		available=available,
		supported=supported,
		reason=reason,
		manifest_url=manifest_url,
		manifest=manifest,
		checked_at=checked_at,
		selected_channel=selected_channel,
	)


def install_launcher_update(
	paths: AppPaths,
	*,
	current_version: str = __version__,
	manifest_url: str | None = None,
	public_key_path: Path | None = None,
	environment: Mapping[str, str] | None = None,
	timeout_seconds: float = 60.0,
	channel: str = STABLE_CHANNEL,
) -> LauncherUpdateResult:
	"""Authenticate, stage, validate, and atomically activate one update."""

	check = check_for_launcher_update(
		current_version=current_version,
		manifest_url=manifest_url,
		public_key_path=public_key_path,
		environment=environment,
		timeout_seconds=min(timeout_seconds, 30.0),
		channel=channel,
	)
	if not check.available:
		return LauncherUpdateResult(
			changed=False,
			current_version=check.current_version,
			installed_version=check.current_version,
			release_directory=None,
			previous_release=None,
			archive=None,
			archive_sha256=None,
			restart_required=False,
			state_recorded=False,
			warning=None,
			selected_channel=check.selected_channel,
			installed_channel=SemanticVersion.parse(check.current_version).channel,
		)
	if not check.supported:
		raise LauncherUpdateError(check.reason)

	from_channel = SemanticVersion.parse(check.current_version).channel
	event_metadata = _launcher_release_metadata(
		current_version=check.current_version,
		target_version=check.manifest.version,
		current_channel=from_channel,
		target_channel=check.manifest.channel,
		selected_channel=check.selected_channel,
		tag=check.manifest.tag,
		manifest_schema=check.manifest.schema,
		manifest_sha256=check.manifest.manifest_sha256,
		rollback=False,
	)
	paths.create_base_directories()
	with _launcher_mutation_lock(paths, "launcher-update"):
		_recover_interrupted_activation(paths)
		_precheck_release_install(paths, check.manifest)
		archive = paths.launcher_update_archive(check.manifest.version)
		_download_archive(
			check.manifest,
			archive,
			timeout_seconds=timeout_seconds,
			event_family="update",
			event_metadata=event_metadata,
		)
		_send_launcher_release_event(
			"update_apply_started",
			metadata=event_metadata,
			operation="launcher_update",
			stage="apply",
		)
		try:
			release_directory = _stage_and_install_release(paths, archive, check.manifest)
			previous = _activate_release(
				paths,
				release_directory,
				telemetry_metadata=event_metadata,
				restore_reason="failed_update",
			)
		except BaseException as exc:
			_send_launcher_release_event(
				"update_apply_failed",
				metadata=event_metadata,
				operation="launcher_update",
				stage="apply",
				severity="error",
				message=str(exc),
			)
			_record_launcher_operation(
				paths,
				from_version=check.current_version,
				to_version=check.manifest.version,
				from_channel=from_channel,
				to_channel=check.manifest.channel,
				selected_channel=check.selected_channel,
				result="failed",
				rollback=False,
				reason="automatic_update",
				tag=check.manifest.tag,
				manifest_schema=check.manifest.schema,
				manifest_sha256=check.manifest.manifest_sha256,
			)
			raise
		_send_launcher_release_event(
			"update_apply_success",
			metadata=event_metadata,
			operation="launcher_update",
			stage="apply",
		)
		state_warning = None
		try:
			_write_json(
				paths.launcher_update_state,
				{
					"schema": 1,
					"updated_at": datetime.now().astimezone().isoformat(),
					"from_version": check.current_version,
					"to_version": check.manifest.version,
					"selected_channel": check.selected_channel,
					"from_channel": from_channel,
					"to_channel": check.manifest.channel,
					"tag": check.manifest.tag,
					"manifest_schema": check.manifest.schema,
					"manifest_sha256": check.manifest.manifest_sha256,
					"manifest_url": check.manifest_url,
					"archive_url": check.manifest.archive_url,
					"archive_sha256": check.manifest.archive_sha256,
					"release_directory": str(release_directory),
					"previous_release": str(previous) if previous else None,
				},
			)
		except OSError as exc:
			# Activation is the durable user-facing operation. A diagnostics-only
			# state record must not strand the dashboard on the old release.
			state_warning = f"launcher updated, but update history could not be saved: {exc}"
		history_recorded = _record_launcher_operation(
			paths,
			from_version=check.current_version,
			to_version=check.manifest.version,
			from_channel=from_channel,
			to_channel=check.manifest.channel,
			selected_channel=check.selected_channel,
			result="success",
			rollback=False,
			reason="automatic_update",
			tag=check.manifest.tag,
			manifest_schema=check.manifest.schema,
			manifest_sha256=check.manifest.manifest_sha256,
		)
		if not history_recorded and state_warning is None:
			state_warning = "launcher updated, but the local operation history could not be saved"
	return LauncherUpdateResult(
		changed=True,
		current_version=check.current_version,
		installed_version=check.manifest.version,
		release_directory=str(release_directory),
		previous_release=str(previous) if previous else None,
		archive=str(archive),
		archive_sha256=check.manifest.archive_sha256,
		restart_required=True,
		state_recorded=state_warning is None,
		warning=state_warning,
		selected_channel=check.selected_channel,
		installed_channel=check.manifest.channel,
		tag=check.manifest.tag,
		manifest_schema=check.manifest.schema,
		manifest_sha256=check.manifest.manifest_sha256,
	)


def verify_launcher_update_channel(
	*,
	current_version: str = __version__,
	manifest_url: str | None = None,
	public_key_path: Path | None = None,
	environment: Mapping[str, str] | None = None,
	timeout_seconds: float = 60.0,
	channel: str = STABLE_CHANNEL,
) -> LauncherUpdateVerification:
	"""Authenticate, download, extract, and import a release without activating it."""

	check = check_for_launcher_update(
		current_version=current_version,
		manifest_url=manifest_url,
		public_key_path=public_key_path,
		environment=environment,
		timeout_seconds=min(timeout_seconds, 30.0),
		channel=channel,
	)
	with tempfile.TemporaryDirectory(prefix="ptlinux-update-verification-") as temporary:
		root = Path(temporary)
		archive = root / f"{UPDATE_ARCHIVE_PREFIX}{check.manifest.version}.tar.gz"
		_download_archive(
			check.manifest,
			archive,
			timeout_seconds=timeout_seconds,
			event_family="update",
			event_metadata=_launcher_release_metadata(
				current_version=current_version,
				target_version=check.manifest.version,
				current_channel=SemanticVersion.parse(current_version).channel,
				target_channel=check.manifest.channel,
				selected_channel=check.selected_channel,
				tag=check.manifest.tag,
				manifest_schema=check.manifest.schema,
				manifest_sha256=check.manifest.manifest_sha256,
				rollback=False,
			),
		)
		staging = root / "release"
		staging.mkdir(mode=0o700)
		_extract_release_archive(archive, staging, check.manifest.version)
		_verify_manifest_files(staging, check.manifest)
		staged_version = _validate_release_directory(staging)
		if staged_version != check.manifest.version:
			raise LauncherUpdateError(
				f"verified launcher version is {staged_version}, "
				f"expected {check.manifest.version}"
			)
		_self_check_release(staging, check.manifest.version)
		_validate_release_directory(staging)
		payload_files = sum(1 for item in staging.rglob("*") if item.is_file())
	return LauncherUpdateVerification(
		current_version=current_version,
		release_version=check.manifest.version,
		manifest_url=check.manifest_url,
		archive_url=check.manifest.archive_url,
		archive_size=check.manifest.archive_size,
		archive_sha256=check.manifest.archive_sha256,
		payload_files=payload_files,
		channel=check.manifest.channel,
		tag=check.manifest.tag,
		manifest_sha256=check.manifest.manifest_sha256,
	)


def launcher_rollback_candidates(
	history: LauncherRemoteHistory,
	*,
	installed_version: str,
	selected_channel: str,
	show_prereleases: bool | None = None,
) -> tuple[LauncherRemoteRelease, ...]:
	"""Resolve valid rollback choices from authenticated release history."""

	try:
		resolver = UpdateResolver(release.resolver_record() for release in history.releases)
		resolved = resolver.rollback_candidates(
			installed_version,
			selected_channel,
			show_prereleases=show_prereleases,
		)
	except ReleaseModelError as exc:
		raise LauncherUpdateError(str(exc)) from exc
	by_version = {release.version: release for release in history.releases}
	return tuple(by_version[release.version] for release in resolved)


def rollback_launcher_update(
	paths: AppPaths,
	*,
	target_version: str | None = None,
	selected_channel: str = STABLE_CHANNEL,
	history_url: str | None = None,
	public_key_path: Path | None = None,
	environment: Mapping[str, str] | None = None,
	timeout_seconds: float = 60.0,
	reason: str = "manual",
) -> LauncherRollbackResult:
	"""Activate a retained or authenticated historical launcher release."""

	try:
		validate_channel(selected_channel)
	except ReleaseModelError as exc:
		raise LauncherUpdateError(str(exc)) from exc
	if reason not in {
		"manual",
		"failed_update",
		"failed_launch",
		"support_recommendation",
		"hotfix_recovery",
	}:
		raise LauncherUpdateError(f"unsupported rollback reason: {reason!r}")
	paths.create_base_directories()
	with _launcher_mutation_lock(paths, "launcher-rollback"):
		_recover_interrupted_activation(paths)
		current_release = _release_from_link(
			paths.launcher_current,
			paths.launcher_releases,
		)
		if current_release is None:
			raise LauncherUpdateError("the current launcher release link is missing")
		current_version = _validate_release_directory(current_release)
		current_semantic = _parse_version(current_version, "current launcher version")
		manifest: UpdateManifest | None = None
		archive: Path | None = None
		release_id: int | None = None
		remote_target = False
		retained_compatibility: tuple[int, int] | None = None
		if target_version is None:
			target_release = _release_from_link(
				paths.launcher_previous,
				paths.launcher_releases,
			)
			if target_release is None:
				raise LauncherUpdateError("no previous launcher release is available")
			target_version = _validate_release_directory(target_release)
			retained_compatibility = _verify_retained_release_integrity(
				paths,
				target_release,
				target_version,
			)
		else:
			try:
				target_semantic = SemanticVersion.parse(target_version)
			except ReleaseModelError as exc:
				raise LauncherUpdateError(f"invalid rollback version: {target_version!r}") from exc
			if target_semantic >= current_semantic:
				raise LauncherUpdateError(
					f"rollback target {target_version} must be older than {current_version}"
				)
			target_release = paths.launcher_releases / target_version
			if target_release.is_dir() and not target_release.is_symlink():
				if _validate_release_directory(target_release) != target_version:
					raise LauncherUpdateError("retained rollback release version does not match")
				retained_compatibility = _verify_retained_release_integrity(
					paths,
					target_release,
					target_version,
				)
			else:
				remote_target = True
		target_semantic = _parse_version(target_version, "rollback target version")
		if target_semantic >= current_semantic:
			raise LauncherUpdateError(
				f"rollback target {target_version} must be older than {current_version}"
			)
		from_channel = current_semantic.channel
		to_channel = target_semantic.channel
		event_metadata = _launcher_release_metadata(
			current_version=current_version,
			target_version=target_version,
			current_channel=from_channel,
			target_channel=to_channel,
			selected_channel=selected_channel,
			tag=manifest.tag if manifest is not None else target_semantic.tag,
			manifest_schema=manifest.schema if manifest is not None else None,
			manifest_sha256=(
				manifest.manifest_sha256 if manifest is not None else None
			),
			rollback=True,
			reason=reason,
			release_id=release_id,
		)
		_send_launcher_release_event(
			"rollback_requested",
			metadata=event_metadata,
			operation="launcher_rollback",
			stage="requested",
		)
		if remote_target:
			history = fetch_launcher_release_history(
				history_url=history_url,
				public_key_path=public_key_path,
				environment=environment,
				timeout_seconds=min(timeout_seconds, 30.0),
			)
			candidates = launcher_rollback_candidates(
				history,
				installed_version=current_version,
				selected_channel=selected_channel,
				show_prereleases=target_semantic.is_prerelease,
			)
			record = next(
				(release for release in candidates if release.version == target_version),
				None,
			)
			if record is None:
				raise LauncherUpdateError(
					f"release {target_version} is not an active rollback candidate"
				)
			release_id = record.release_id
			fetched_manifest = _fetch_historical_manifest(
				record,
				public_key_path or configured_public_key_path(environment),
				timeout_seconds=min(timeout_seconds, 30.0),
			)
			_validate_rollback_config_compatibility(paths, fetched_manifest)
			_precheck_release_install(paths, fetched_manifest)
			manifest = fetched_manifest
			archive = paths.launcher_update_archive(fetched_manifest.version)
			event_metadata = _launcher_release_metadata(
				current_version=current_version,
				target_version=target_version,
				current_channel=from_channel,
				target_channel=to_channel,
				selected_channel=selected_channel,
				tag=fetched_manifest.tag,
				manifest_schema=fetched_manifest.schema,
				manifest_sha256=fetched_manifest.manifest_sha256,
				rollback=True,
				reason=reason,
				release_id=release_id,
			)
		if manifest is not None and archive is not None:
			_download_archive(
				manifest,
				archive,
				timeout_seconds=timeout_seconds,
				event_family="rollback",
				event_metadata=event_metadata,
			)
		_send_launcher_release_event(
			"rollback_apply_started",
			metadata=event_metadata,
			operation="launcher_rollback",
			stage="apply",
		)
		try:
			if manifest is not None and archive is not None:
				target_release = _stage_and_install_release(paths, archive, manifest)
			if current_release == target_release.resolve():
				raise LauncherUpdateError("current and rollback releases are identical")
			if manifest is not None:
				_validate_rollback_config_compatibility(paths, manifest)
			elif retained_compatibility is not None:
				config_schema, minimum_config_schema = retained_compatibility
				_validate_config_schema_range(
					_current_config_schema(paths),
					minimum_config_schema,
					config_schema,
					f"launcher {target_version}",
				)
			else:
				_validate_rollback_release_compatibility(paths, target_release)
			previous = _activate_release(
				paths,
				target_release,
				telemetry_metadata=event_metadata,
				restore_reason=reason,
			)
			if previous is None:
				raise LauncherUpdateError("the current launcher release disappeared")
		except BaseException as exc:
			failed_tag = event_metadata.get("release_tag")
			if not isinstance(failed_tag, str):
				failed_tag = None
			_send_launcher_release_event(
				"rollback_apply_failed",
				metadata=event_metadata,
				operation="launcher_rollback",
				stage="apply",
				severity="error",
				message=str(exc),
			)
			_record_launcher_operation(
				paths,
				from_version=current_version,
				to_version=target_version,
				from_channel=from_channel,
				to_channel=to_channel,
				selected_channel=selected_channel,
				result="failed",
				rollback=True,
				reason=reason,
				tag=failed_tag,
				manifest_schema=manifest.schema if manifest is not None else None,
				manifest_sha256=(
					manifest.manifest_sha256 if manifest is not None else None
				),
			)
			raise
		_send_launcher_release_event(
			"rollback_apply_success",
			metadata=event_metadata,
			operation="launcher_rollback",
			stage="apply",
		)
		state_warning = None
		try:
			_write_json(
				paths.launcher_update_state,
				{
					"schema": 1,
					"rolled_back_at": datetime.now().astimezone().isoformat(),
					"from_version": current_version,
					"to_version": target_version,
					"selected_channel": selected_channel,
					"from_channel": from_channel,
					"to_channel": to_channel,
					"reason": reason,
					"rollback": True,
					"release_directory": str(target_release),
					"previous_release": str(previous),
				},
			)
		except OSError as exc:
			# The symlink transaction is already durable. Diagnostics must not make
			# a completed rollback appear to have failed or encourage a retry.
			state_warning = f"launcher rolled back, but update history could not be saved: {exc}"
		history_recorded = _record_launcher_operation(
			paths,
			from_version=current_version,
			to_version=target_version,
			from_channel=from_channel,
			to_channel=to_channel,
			selected_channel=selected_channel,
			result="success",
			rollback=True,
			reason=reason,
			tag=manifest.tag if manifest is not None else target_semantic.tag,
			manifest_schema=manifest.schema if manifest is not None else None,
			manifest_sha256=(
				manifest.manifest_sha256 if manifest is not None else None
			),
		)
		if not history_recorded and state_warning is None:
			state_warning = "launcher rolled back, but the local operation history could not be saved"
	return LauncherRollbackResult(
		from_version=current_version,
		to_version=target_version,
		current_release=str(target_release),
		previous_release=str(previous),
		restart_required=True,
		from_channel=from_channel,
		to_channel=to_channel,
		state_recorded=state_warning is None,
		warning=state_warning,
	)


def inspect_launcher_update_history(paths: AppPaths) -> LauncherUpdateHistory:
	"""Return validated retained releases without trusting directory names alone."""

	current = _release_from_link(paths.launcher_current, paths.launcher_releases)
	previous = _release_from_link(paths.launcher_previous, paths.launcher_releases)
	try:
		candidates = tuple(paths.launcher_releases.iterdir())
	except FileNotFoundError:
		candidates = ()
	except OSError as exc:
		raise LauncherUpdateError(f"cannot inspect installed launcher releases: {exc}") from exc
	releases: list[LauncherInstalledRelease] = []
	for candidate in candidates:
		try:
			SemanticVersion.parse(candidate.name)
		except ReleaseModelError:
			continue
		try:
			version = _validate_release_directory(candidate)
			installed_at = datetime.fromtimestamp(
				candidate.stat().st_mtime
			).astimezone().isoformat(timespec="seconds")
		except (LauncherUpdateError, OSError, ValueError, OverflowError):
			continue
		if version != candidate.name:
			continue
		resolved = candidate.resolve()
		releases.append(
			LauncherInstalledRelease(
				version=version,
				installed_at=installed_at,
				directory=str(resolved),
				active=resolved == current,
				previous=resolved == previous,
				channel=SemanticVersion.parse(version).channel,
			)
		)
	releases.sort(
		key=lambda release: _parse_version(release.version, "launcher release version"),
		reverse=True,
	)
	return LauncherUpdateHistory(
		current_version=_release_version(releases, active=True),
		previous_version=_release_version(releases, previous=True),
		releases=tuple(releases),
	)


def parse_launcher_release_history(
	data: bytes,
	*,
	history_url: str,
) -> LauncherRemoteHistory:
	"""Strictly parse an already authenticated remote release catalog."""

	if not data or len(data) > MAX_HISTORY_BYTES:
		raise LauncherUpdateError("release history has an invalid size")
	_require_https_url(history_url, "release history URL")
	try:
		value = json.loads(
			data.decode("utf-8"),
			object_pairs_hook=_unique_json_object,
		)
	except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc:
		raise LauncherUpdateError(f"release history is invalid JSON: {exc}") from exc
	if not isinstance(value, dict):
		raise LauncherUpdateError("release history must contain a JSON object")
	if value.get("application") != UPDATE_APPLICATION:
		raise LauncherUpdateError(
			f"release history application must be {UPDATE_APPLICATION!r}"
		)
	if _required_integer(value, "schema") != RELEASE_HISTORY_SCHEMA:
		raise LauncherUpdateError("unsupported release history schema")
	generated_at = _required_string(value, "generated_at")
	_parse_published_at(generated_at)
	releases_value = value.get("releases")
	if not isinstance(releases_value, list) or len(releases_value) > MAX_RELEASE_HISTORY:
		raise LauncherUpdateError("release history releases must be a bounded array")
	releases: list[LauncherRemoteRelease] = []
	seen: set[SemanticVersion] = set()
	last_version: SemanticVersion | None = None
	for item in releases_value:
		if not isinstance(item, dict):
			raise LauncherUpdateError("release history entry must be an object")
		version = _required_string(item, "version")
		channel = _required_string(item, "channel")
		tag = _required_string(item, "tag")
		try:
			parsed = validate_channel_version(channel, version)
		except ReleaseModelError as exc:
			raise LauncherUpdateError(f"invalid release history entry: {exc}") from exc
		if tag != parsed.tag:
			raise LauncherUpdateError(
				f"release history tag {tag!r} does not match version {version}"
			)
		if parsed in seen:
			raise LauncherUpdateError(f"duplicate release history version: {version}")
		if last_version is not None and parsed >= last_version:
			raise LauncherUpdateError(
				"release history entries must be sorted by descending semantic version"
			)
		seen.add(parsed)
		last_version = parsed
		status = _required_string(item, "status")
		if status not in SUPPORTED_RELEASE_STATUSES:
			raise LauncherUpdateError(f"unsupported release status: {status!r}")
		manifest_url = _required_string(item, "manifest_url")
		manifest_signature_url = _required_string(item, "manifest_signature_url")
		release_date = _required_string(item, "release_date")
		release_notes_url = _required_string(item, "release_notes_url")
		for url, label in (
			(manifest_url, "historical manifest URL"),
			(manifest_signature_url, "historical manifest signature URL"),
			(release_notes_url, "historical release notes URL"),
		):
			_require_https_url(url, label)
		if _resource_url(manifest_signature_url) != _resource_url(
			_signature_url(manifest_url)
		):
			raise LauncherUpdateError(
				f"release history signature URL does not match manifest {version}"
			)
		_parse_published_at(release_date)
		title = _required_string(item, "title")
		release_id_value = item.get("release_id")
		if release_id_value is not None and (
			isinstance(release_id_value, bool)
			or not isinstance(release_id_value, int)
			or release_id_value <= 0
		):
			raise LauncherUpdateError("release history release_id must be positive or null")
		config_schema, minimum_config_schema = _parse_history_compatibility(
			item.get("compatibility")
		)
		releases.append(
			LauncherRemoteRelease(
				version=version,
				channel=channel,
				tag=tag,
				manifest_url=manifest_url,
				manifest_signature_url=manifest_signature_url,
				release_date=release_date,
				release_notes_url=release_notes_url,
				status=status,
				title=title,
				release_id=release_id_value,
				config_schema=config_schema,
				minimum_config_schema=minimum_config_schema,
			)
		)
	return LauncherRemoteHistory(
		generated_at=generated_at,
		history_url=history_url,
		releases=tuple(releases),
		history_sha256=hashlib.sha256(data).hexdigest(),
	)


def _parse_history_compatibility(value: object) -> tuple[int | None, int | None]:
	if value is None:
		return None, None
	if not isinstance(value, dict):
		raise LauncherUpdateError("release history compatibility must be an object")
	config_schema = _optional_positive_integer(value, "config_schema")
	minimum_config_schema = _optional_positive_integer(
		value,
		"minimum_config_schema",
	)
	if (config_schema is None) != (minimum_config_schema is None):
		raise LauncherUpdateError(
			"release history compatibility must include both config schema bounds"
		)
	if (
		config_schema is not None
		and minimum_config_schema is not None
		and minimum_config_schema > config_schema
	):
		raise LauncherUpdateError(
			"release history minimum_config_schema exceeds config_schema"
		)
	return config_schema, minimum_config_schema


def parse_update_manifest(data: bytes) -> UpdateManifest:
	"""Strictly parse one already-authenticated manifest."""

	if not data or len(data) > MAX_MANIFEST_BYTES:
		raise LauncherUpdateError("update manifest has an invalid size")
	try:
		value = json.loads(
			data.decode("utf-8"),
			object_pairs_hook=_unique_json_object,
		)
	except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc:
		raise LauncherUpdateError(f"update manifest is invalid JSON: {exc}") from exc
	if not isinstance(value, dict):
		raise LauncherUpdateError("update manifest must contain a JSON object")
	application = value.get("application")
	if application is not None and application != UPDATE_APPLICATION:
		raise LauncherUpdateError(
			f"update manifest application must be {UPDATE_APPLICATION!r}"
		)
	schema = _required_integer(value, "schema")
	if schema != 1:
		raise LauncherUpdateError(f"unsupported update manifest schema: {schema}")
	channel = _required_string(value, "channel")
	version = _required_string(value, "version")
	minimum = _required_string(value, "minimum_updater_version")
	try:
		parsed_version = validate_channel_version(channel, version)
	except ReleaseModelError as exc:
		raise LauncherUpdateError(f"invalid update manifest release: {exc}") from exc
	_parse_version(minimum, "minimum updater version")
	archive_url = _required_string(value, "archive_url")
	_require_https_url(archive_url, "update archive URL")
	archive_sha256 = _required_string(value, "archive_sha256")
	if _SHA256_PATTERN.fullmatch(archive_sha256) is None:
		raise LauncherUpdateError("archive_sha256 must be a lowercase SHA-256 digest")
	archive_size = _required_integer(value, "archive_size")
	if archive_size <= 0 or archive_size > MAX_ARCHIVE_BYTES:
		raise LauncherUpdateError(
			f"update archive size must be between 1 and {MAX_ARCHIVE_BYTES} bytes"
		)
	published_at = _required_string(value, "published_at")
	_parse_published_at(published_at)
	notes_value = value.get("notes_url")
	if notes_value is not None and not isinstance(notes_value, str):
		raise LauncherUpdateError("notes_url must be a string or null")
	if notes_value:
		_require_https_url(notes_value, "release notes URL")
	tag_value = value.get("tag", parsed_version.tag)
	if not isinstance(tag_value, str) or tag_value != parsed_version.tag:
		raise LauncherUpdateError("manifest tag does not match its version")
	commit_value = value.get("commit")
	if commit_value is not None and (
		not isinstance(commit_value, str)
		or re.fullmatch(r"[0-9a-f]{7,64}", commit_value) is None
	):
		raise LauncherUpdateError("manifest commit must be a lowercase Git object ID")
	status_value = value.get("status", ACTIVE_STATUS)
	if not isinstance(status_value, str) or status_value not in SUPPORTED_RELEASE_STATUSES:
		raise LauncherUpdateError(f"unsupported release status: {status_value!r}")
	title_value: str | None = None
	summary_value: str | None = None
	notes_markdown_value: str | None = None
	release_value = value.get("release")
	if release_value is not None:
		if not isinstance(release_value, dict):
			raise LauncherUpdateError("manifest release field must be an object")
		title_value = _required_string(release_value, "title")
		summary_value = _required_string(release_value, "summary")
		notes_markdown_candidate = release_value.get("notes_markdown")
		if notes_markdown_candidate is not None:
			if (
				not isinstance(notes_markdown_candidate, str)
				or not notes_markdown_candidate
				or "\x00" in notes_markdown_candidate
				or len(notes_markdown_candidate.encode("utf-8"))
				> MAX_EMBEDDED_RELEASE_NOTES_BYTES
			):
				raise LauncherUpdateError(
					"manifest release.notes_markdown is invalid or too large"
				)
			notes_markdown_value = notes_markdown_candidate
		release_notes_value = _required_string(release_value, "release_notes_url")
		_require_https_url(release_notes_value, "release notes URL")
		if notes_value and notes_value != release_notes_value:
			raise LauncherUpdateError("manifest release notes URLs do not match")
		notes_value = release_notes_value
	archive_name = PurePosixPath(urlsplit(archive_url).path).name
	archive_value = value.get("archive")
	if archive_value is not None:
		if not isinstance(archive_value, dict):
			raise LauncherUpdateError("manifest archive field must be an object")
		archive_name = _required_string(archive_value, "name")
		if PurePosixPath(archive_name).name != archive_name:
			raise LauncherUpdateError("manifest archive name must be a filename")
		if PurePosixPath(urlsplit(archive_url).path).name != archive_name:
			raise LauncherUpdateError("manifest archive name does not match its URL")
		if (
			_required_string(archive_value, "url") != archive_url
			or _required_string(archive_value, "sha256") != archive_sha256
			or _required_integer(archive_value, "size") != archive_size
		):
			raise LauncherUpdateError("manifest archive fields do not match legacy metadata")
	config_schema = 1
	minimum_config_schema = 1
	compatibility_value = value.get("compatibility")
	if compatibility_value is not None:
		if not isinstance(compatibility_value, dict):
			raise LauncherUpdateError("manifest compatibility field must be an object")
		config_schema = _required_positive_integer(
			compatibility_value,
			"config_schema",
		)
		minimum_config_schema = _required_positive_integer(
			compatibility_value,
			"minimum_config_schema",
		)
		if minimum_config_schema > config_schema:
			raise LauncherUpdateError(
				"manifest minimum_config_schema exceeds config_schema"
			)
		compatibility_manifest_schema = compatibility_value.get("manifest_schema")
		if compatibility_manifest_schema is not None and compatibility_manifest_schema != schema:
			raise LauncherUpdateError(
				"manifest compatibility schema does not match its schema"
			)
		compatibility_minimum = compatibility_value.get("minimum_updater_version")
		if compatibility_minimum is not None and compatibility_minimum != minimum:
			raise LauncherUpdateError(
				"manifest compatibility minimum updater version does not match"
			)
	files = _parse_manifest_files(value.get("files"))
	return UpdateManifest(
		schema=schema,
		channel=channel,
		version=version,
		minimum_updater_version=minimum,
		archive_url=archive_url,
		archive_sha256=archive_sha256,
		archive_size=archive_size,
		published_at=published_at,
		notes_url=notes_value or None,
		tag=tag_value,
		commit=commit_value,
		status=status_value,
		title=title_value,
		summary=summary_value,
		notes_markdown=notes_markdown_value,
		archive_name=archive_name,
		files=files,
		manifest_sha256=hashlib.sha256(data).hexdigest(),
		config_schema=config_schema,
		minimum_config_schema=minimum_config_schema,
	)


def _parse_manifest_files(value: object) -> tuple[ManifestFile, ...]:
	if value is None:
		return ()
	if not isinstance(value, list) or len(value) > MAX_ARCHIVE_MEMBERS:
		raise LauncherUpdateError("manifest files must be a bounded array")
	files: list[ManifestFile] = []
	last_path = ""
	for item in value:
		if not isinstance(item, dict):
			raise LauncherUpdateError("manifest file entry must be an object")
		path_value = _required_string(item, "path")
		raw_parts = path_value.split("/")
		path = PurePosixPath(path_value)
		if (
			"\\" in path_value
			or "\x00" in path_value
			or any(part in {"", ".", ".."} for part in raw_parts)
			or path.is_absolute()
			or not path.parts
			or path.parts[0] not in _ALLOWED_RELEASE_ROOTS
		):
			raise LauncherUpdateError(f"unsafe manifest file path: {path_value!r}")
		if path_value <= last_path:
			raise LauncherUpdateError("manifest files must be uniquely sorted by path")
		size = _required_integer(item, "size")
		if size < 0 or size > MAX_EXTRACTED_BYTES:
			raise LauncherUpdateError(f"invalid manifest file size: {path_value}")
		digest = _required_string(item, "sha256")
		if _SHA256_PATTERN.fullmatch(digest) is None:
			raise LauncherUpdateError(f"invalid manifest file SHA-256: {path_value}")
		files.append(ManifestFile(path_value, size, digest))
		last_path = path_value
	return tuple(files)


def _fetch_small_file(
	url: str,
	*,
	maximum_bytes: int,
	timeout_seconds: float,
	label: str,
) -> bytes:
	request = Request(
		url,
		headers={
			"User-Agent": USER_AGENT,
			"Accept": "application/octet-stream, application/json",
			"Accept-Encoding": "identity",
			"Cache-Control": "no-cache",
			"Pragma": "no-cache",
		},
		method="GET",
	)
	try:
		response = urlopen(request, timeout=timeout_seconds)
	except HTTPError as exc:
		status = exc.code
		exc.close()
		raise LauncherUpdateError(f"{label} request failed: HTTP {status}") from exc
	except (URLError, OSError) as exc:
		raise LauncherUpdateError(f"{label} request failed: {exc}") from exc
	with response:
		final_url = response.geturl()
		_require_https_url(final_url, f"final {label} URL")
		if _resource_url(final_url) != _resource_url(url):
			raise LauncherUpdateError(
				f"{label} redirected unexpectedly to {final_url}; "
				"the launcher update channel must be publicly readable"
			)
		length = _content_length(response.headers.get("Content-Length"), label)
		if length is not None and length > maximum_bytes:
			raise LauncherUpdateError(f"{label} exceeds its size limit")
		data = response.read(maximum_bytes + 1)
	if len(data) > maximum_bytes:
		raise LauncherUpdateError(f"{label} exceeds its size limit")
	if length is not None and len(data) != length:
		raise LauncherUpdateError(f"{label} response was incomplete")
	if not data:
		raise LauncherUpdateError(f"{label} response was empty")
	return data


def _verify_manifest_signature(
	manifest: bytes,
	signature: bytes,
	public_key_path: Path,
) -> None:
	if not signature or len(signature) > MAX_SIGNATURE_BYTES:
		raise LauncherUpdateError("update manifest signature has an invalid size")
	key = public_key_path.expanduser().resolve()
	if not key.is_file():
		raise LauncherUpdateError(f"launcher update public key is missing: {key}")
	openssl = shutil.which("openssl")
	if not openssl:
		raise LauncherUpdateError("OpenSSL is required to verify launcher updates")
	with tempfile.TemporaryDirectory(prefix="ptlinux-update-signature-") as temporary:
		root = Path(temporary)
		manifest_path = root / "manifest.json"
		signature_path = root / "manifest.json.sig"
		manifest_path.write_bytes(manifest)
		signature_path.write_bytes(signature)
		commands = (
			[
				openssl,
				"dgst",
				"-sha256",
				"-verify",
				str(key),
				"-signature",
				str(signature_path),
				str(manifest_path),
			],
			[
				openssl,
				"pkeyutl",
				"-verify",
				"-rawin",
				"-pubin",
				"-inkey",
				str(key),
				"-sigfile",
				str(signature_path),
				"-in",
				str(manifest_path),
			],
		)
		for command in commands:
			try:
				result = subprocess.run(
					command,
					capture_output=True,
					text=True,
					encoding="utf-8",
					errors="replace",
					timeout=10,
					check=False,
				)
			except (OSError, subprocess.TimeoutExpired):
				continue
			if result.returncode == 0:
				return
	raise LauncherUpdateError("launcher update manifest signature is invalid")


def _launcher_release_metadata(
	*,
	current_version: str,
	target_version: str,
	current_channel: str,
	target_channel: str,
	selected_channel: str,
	tag: str | None,
	manifest_schema: int | None,
	manifest_sha256: str | None,
	rollback: bool,
	reason: str | None = None,
	release_id: int | None = None,
) -> dict[str, object]:
	metadata: dict[str, object] = {
		"current_version": current_version,
		"target_version": target_version,
		"current_channel": current_channel,
		"target_channel": target_channel,
		"selected_update_channel": selected_channel,
		"release_tag": tag,
		"manifest_version": manifest_schema,
		"manifest_hash": manifest_sha256,
		"rollback": rollback,
	}
	if reason is not None:
		metadata["reason"] = reason
	if release_id is not None:
		metadata["release_id"] = release_id
	if rollback:
		metadata.update(
			{
				"from_version": current_version,
				"to_version": target_version,
				"from_channel": current_channel,
				"to_channel": target_channel,
			}
		)
	return metadata


def _send_launcher_release_event(
	event: str,
	*,
	metadata: Mapping[str, object],
	operation: str,
	stage: str,
	severity: str = "info",
	message: str | None = None,
) -> None:
	telemetry = current_telemetry()
	if telemetry is not None:
		telemetry.send_event(
			event,
			severity=severity,
			message=message,
			metadata=metadata,
			context={"operation": operation, "stage": stage},
		)


def _download_archive(
	manifest: UpdateManifest,
	destination: Path,
	*,
	timeout_seconds: float,
	event_family: str | None = None,
	event_metadata: Mapping[str, object] | None = None,
) -> None:
	if destination.is_file() and _archive_matches(destination, manifest):
		return
	telemetry = current_telemetry()
	download_started_at = time.monotonic()
	operation = "launcher_rollback" if event_family == "rollback" else "launcher_update"
	started_event = (
		f"{event_family}_download_started"
		if event_family is not None
		else "download_started"
	)
	success_event = (
		f"{event_family}_download_success"
		if event_family is not None
		else "download_completed"
	)
	failed_event = (
		f"{event_family}_download_failed"
		if event_family is not None
		else "download_failed"
	)
	base_metadata = {
		"resource_type": "launcher_update",
		"content_bytes": manifest.archive_size,
		"mirror_id": "launcher_update_channel",
		**dict(event_metadata or {}),
	}
	if telemetry is not None:
		telemetry.send_event(
			started_event,
			metadata=base_metadata,
			context={"operation": operation, "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)
	request = Request(
		manifest.archive_url,
		headers={
			"User-Agent": USER_AGENT,
			"Accept": "application/gzip, application/octet-stream",
			"Accept-Encoding": "identity",
		},
		method="GET",
	)
	written = 0
	try:
		try:
			response = urlopen(request, timeout=timeout_seconds)
		except HTTPError as exc:
			status = exc.code
			exc.close()
			raise LauncherUpdateError(
				f"launcher archive request failed: HTTP {status}"
			) from exc
		except (URLError, OSError) as exc:
			raise LauncherUpdateError(f"launcher archive request failed: {exc}") from exc
		with response:
			_require_https_url(response.geturl(), "final update archive URL")
			length = _content_length(
				response.headers.get("Content-Length"),
				"launcher archive",
			)
			if length is not None and length != manifest.archive_size:
				raise LauncherUpdateError(
					"launcher archive Content-Length does not match the signed manifest"
				)
			digest = hashlib.sha256()
			with temporary.open("xb") as handle:
				while True:
					chunk = response.read(READ_SIZE)
					if not chunk:
						break
					written += len(chunk)
					if written > manifest.archive_size or written > MAX_ARCHIVE_BYTES:
						raise LauncherUpdateError("launcher archive exceeded its signed size")
					digest.update(chunk)
					handle.write(chunk)
				handle.flush()
				os.fsync(handle.fileno())
		if written != manifest.archive_size:
			raise LauncherUpdateError(
				f"launcher archive was incomplete: {written}/{manifest.archive_size} bytes"
			)
		if digest.hexdigest() != manifest.archive_sha256:
			raise LauncherUpdateError("launcher archive SHA-256 does not match the manifest")
		temporary.replace(destination)
		_fsync_directory(destination.parent)
	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(
				failed_event,
				severity="error",
				message=str(exc),
				metadata={
					**base_metadata,
					"download_bytes": written,
					"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),
				},
				context={"operation": operation, "stage": "download"},
			)
		raise
	duration = max(0.0, time.monotonic() - download_started_at)
	if telemetry is not None:
		telemetry.send_event(
			success_event,
			metadata={
				**base_metadata,
				"download_bytes": written,
				"duration_ms": round(duration * 1000),
				"average_mbps": _average_mbps(written, duration),
				"retry_count": 0,
				"resume_used": False,
				"http_status": 200,
			},
			context={"operation": operation, "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(
	paths: AppPaths,
	archive: Path,
	manifest: UpdateManifest,
) -> Path:
	if paths.app_home.is_symlink():
		raise LauncherUpdateError(f"launcher application home cannot be a link: {paths.app_home}")
	paths.app_home.mkdir(parents=True, exist_ok=True, mode=0o755)
	paths.app_home.chmod(0o755)
	if paths.launcher_releases.is_symlink():
		raise LauncherUpdateError(
			f"launcher release directory cannot be a link: {paths.launcher_releases}"
		)
	paths.launcher_releases.mkdir(parents=True, exist_ok=True, mode=0o755)
	paths.launcher_releases.chmod(0o755)
	staging = paths.launcher_releases / f".{manifest.version}.{os.getpid()}.staging"
	if staging.exists() or staging.is_symlink():
		raise LauncherUpdateError(f"launcher staging path already exists: {staging}")
	staging.mkdir(mode=0o700)
	try:
		_extract_release_archive(archive, staging, manifest.version)
		_verify_manifest_files(staging, manifest)
		staged_version = _validate_release_directory(staging)
		if staged_version != manifest.version:
			raise LauncherUpdateError(
				f"staged launcher version is {staged_version}, expected {manifest.version}"
			)
		_self_check_release(staging, manifest.version)
		_validate_release_directory(staging)
		staging.chmod(0o755)
		destination = paths.launcher_releases / manifest.version
		if destination.exists() or destination.is_symlink():
			if not destination.is_dir() or destination.is_symlink():
				raise LauncherUpdateError(
					f"launcher release destination is not a directory: {destination}"
				)
			_validate_release_directory(destination)
			if _tree_digest(destination) != _tree_digest(staging):
				raise LauncherUpdateError(
					f"launcher release {manifest.version} already exists with different contents"
				)
			shutil.rmtree(staging)
		else:
			staging.replace(destination)
			_fsync_directory(paths.launcher_releases)
		_write_release_integrity_record(paths, destination, manifest)
		return destination
	except BaseException:
		if staging.exists() and staging.is_dir() and not staging.is_symlink():
			shutil.rmtree(staging)
		raise


def _verify_manifest_files(release: Path, manifest: UpdateManifest) -> None:
	_verify_file_inventory(release, manifest.files)


def _verify_file_inventory(
	release: Path,
	files: tuple[ManifestFile, ...],
) -> None:
	if not files:
		return
	expected = {entry.path: entry for entry in files}
	actual: set[str] = set()
	for path in release.rglob("*"):
		if not path.is_file():
			continue
		relative = path.relative_to(release).as_posix()
		actual.add(relative)
		entry = expected.get(relative)
		if entry is None:
			raise LauncherUpdateError(
				f"launcher archive file is absent from its signed manifest: {relative}"
			)
		if path.stat().st_size != entry.size or _sha256_file(path) != entry.sha256:
			raise LauncherUpdateError(
				f"launcher archive file does not match its signed manifest: {relative}"
			)
	missing = sorted(set(expected).difference(actual))
	if missing:
		raise LauncherUpdateError(
			"launcher archive is missing signed manifest files: " + ", ".join(missing)
		)


def _write_release_integrity_record(
	paths: AppPaths,
	release: Path,
	manifest: UpdateManifest,
) -> None:
	if not manifest.files:
		return
	directory = paths.launcher_release_integrity
	if directory.is_symlink():
		raise LauncherUpdateError(
			f"launcher release-integrity directory cannot be a link: {directory}"
		)
	directory.mkdir(parents=True, exist_ok=True, mode=0o700)
	directory.chmod(0o700)
	if release.resolve() != (paths.launcher_releases / manifest.version).resolve():
		raise LauncherUpdateError("launcher integrity record release path is inconsistent")
	_write_json(
		directory / f"{manifest.version}.json",
		{
			"schema": 1,
			"version": manifest.version,
			"channel": manifest.channel,
			"tag": manifest.tag or SemanticVersion.parse(manifest.version).tag,
			"manifest_schema": manifest.schema,
			"manifest_sha256": manifest.manifest_sha256,
			"config_schema": manifest.config_schema,
			"minimum_config_schema": manifest.minimum_config_schema,
			"files": [asdict(entry) for entry in manifest.files],
		},
	)


def _verify_retained_release_integrity(
	paths: AppPaths,
	release: Path,
	version: str,
) -> tuple[int, int] | None:
	"""Verify a retained release against its install-time signed file inventory."""

	record = paths.launcher_release_integrity / f"{version}.json"
	try:
		if record.is_symlink():
			raise LauncherUpdateError(f"launcher integrity record cannot be a link: {record}")
		data = record.read_bytes()
	except FileNotFoundError:
		# Releases installed by an older launcher predate integrity persistence.
		return None
	except OSError as exc:
		raise LauncherUpdateError(f"cannot read launcher integrity record: {exc}") from exc
	if not data or len(data) > MAX_MANIFEST_BYTES:
		raise LauncherUpdateError("launcher integrity record has an invalid size")
	try:
		value = json.loads(data.decode("utf-8"), object_pairs_hook=_unique_json_object)
	except (UnicodeError, json.JSONDecodeError, ValueError) as exc:
		raise LauncherUpdateError(f"launcher integrity record is invalid: {exc}") from exc
	if not isinstance(value, dict) or value.get("schema") != 1:
		raise LauncherUpdateError("launcher integrity record has an unsupported schema")
	semantic = _parse_version(version, "retained launcher version")
	if (
		value.get("version") != version
		or value.get("channel") != semantic.channel
		or value.get("tag") != semantic.tag
		or value.get("manifest_schema") != 1
	):
		raise LauncherUpdateError("launcher integrity record identity does not match release")
	manifest_sha256 = value.get("manifest_sha256")
	if manifest_sha256 is not None and (
		not isinstance(manifest_sha256, str)
		or _SHA256_PATTERN.fullmatch(manifest_sha256) is None
	):
		raise LauncherUpdateError("launcher integrity record manifest hash is invalid")
	config_schema = _required_positive_integer(value, "config_schema")
	minimum_config_schema = _required_positive_integer(
		value,
		"minimum_config_schema",
	)
	if minimum_config_schema > config_schema:
		raise LauncherUpdateError(
			"launcher integrity record has invalid config compatibility"
		)
	files = _parse_manifest_files(value.get("files"))
	if not files:
		raise LauncherUpdateError("launcher integrity record has no managed files")
	_verify_file_inventory(release, files)
	return config_schema, minimum_config_schema


def _extract_release_archive(archive: Path, destination: Path, version: str) -> None:
	expected_root = f"{UPDATE_ARCHIVE_PREFIX}{version}"
	archive_layout: str | None = None
	seen: set[PurePosixPath] = set()
	total_bytes = 0
	try:
		bundle = tarfile.open(archive, "r:gz")
	except (OSError, tarfile.TarError) as exc:
		raise LauncherUpdateError(f"launcher archive cannot be opened: {exc}") from exc
	with bundle:
		members = bundle.getmembers()
		if not members or len(members) > MAX_ARCHIVE_MEMBERS:
			raise LauncherUpdateError("launcher archive has an invalid member count")
		for member in members:
			name = member.name[:-1] if member.isdir() and member.name.endswith("/") else member.name
			if not name or "\\" in name or "\x00" in name:
				raise LauncherUpdateError(f"unsafe launcher archive path: {member.name!r}")
			raw_parts = name.split("/")
			if any(part in {"", ".", ".."} for part in raw_parts):
				raise LauncherUpdateError(f"unsafe launcher archive path: {member.name!r}")
			path = PurePosixPath(name)
			if path.is_absolute() or not path.parts:
				raise LauncherUpdateError(
					f"unsafe launcher archive path: {member.name!r}"
				)
			if path.parts[0] == expected_root:
				member_layout = "wrapped"
			elif path.parts[0] in _ALLOWED_RELEASE_ROOTS:
				member_layout = "flat"
			else:
				raise LauncherUpdateError(
					f"unexpected launcher archive path: {member.name!r}"
				)
			if archive_layout is None:
				archive_layout = member_layout
			elif archive_layout != member_layout:
				raise LauncherUpdateError(
					"launcher archive mixes flat and legacy wrapped paths"
				)
			relative = (
				PurePosixPath(*path.parts[1:])
				if archive_layout == "wrapped"
				else path
			)
			if not relative.parts:
				if not member.isdir():
					raise LauncherUpdateError("launcher archive root must be a directory")
				continue
			if relative.parts[0] not in _ALLOWED_RELEASE_ROOTS:
				raise LauncherUpdateError(
					f"unexpected launcher payload path: {relative.as_posix()}"
				)
			if relative in seen:
				raise LauncherUpdateError(
					f"duplicate launcher archive path: {relative.as_posix()}"
				)
			seen.add(relative)
			if not member.isdir() and not member.isfile():
				raise LauncherUpdateError(
					f"unsupported launcher archive member: {member.name!r}"
				)
			if (
				relative.parts[0] == "release.json"
				and (len(relative.parts) != 1 or not member.isfile())
			):
				raise LauncherUpdateError(
					f"unexpected launcher metadata path: {member.name!r}"
				)
			if (
				len(relative.parts) == 1
				and relative.parts[0] in {"ptlinux", "ptdiag"}
				and not member.isdir()
			):
				raise LauncherUpdateError(
					f"launcher package root must be a directory: {member.name!r}"
				)
			if member.mode & 0o022:
				raise LauncherUpdateError(
					f"launcher archive member is group/world writable: {member.name!r}"
				)
			permissions = member.mode & 0o777
			if member.isfile() and permissions not in {0o644, 0o755}:
				raise LauncherUpdateError(
					f"launcher archive member has unsupported permissions: {member.name!r}"
				)
			target = destination.joinpath(*relative.parts)
			if member.isdir():
				target.mkdir(parents=True, exist_ok=True, mode=0o755)
				continue
			if member.size < 0:
				raise LauncherUpdateError("launcher archive contains a negative file size")
			total_bytes += member.size
			if total_bytes > MAX_EXTRACTED_BYTES:
				raise LauncherUpdateError("launcher archive exceeds its extraction limit")
			target.parent.mkdir(parents=True, exist_ok=True, mode=0o755)
			source = bundle.extractfile(member)
			if source is None:
				raise LauncherUpdateError(f"cannot read launcher payload: {member.name!r}")
			written = 0
			with source, target.open("xb") as handle:
				while True:
					chunk = source.read(READ_SIZE)
					if not chunk:
						break
					written += len(chunk)
					if written > member.size:
						raise LauncherUpdateError(
							f"launcher payload exceeded its declared size: {member.name!r}"
						)
					handle.write(chunk)
			if written != member.size:
				raise LauncherUpdateError(
					f"launcher payload was incomplete: {member.name!r}"
				)
			target.chmod(permissions)


def _validate_release_directory(release: Path) -> str:
	if not release.is_dir() or release.is_symlink():
		raise LauncherUpdateError(f"launcher release directory is invalid: {release}")
	for item in (release, *release.rglob("*")):
		if item.is_symlink():
			raise LauncherUpdateError(f"launcher release contains a symbolic link: {item}")
		try:
			mode = item.stat().st_mode
		except OSError as exc:
			raise LauncherUpdateError(f"cannot inspect launcher release path {item}: {exc}") from exc
		if mode & 0o022:
			raise LauncherUpdateError(
				f"launcher release path is group/world writable: {item}"
			)
		if not stat.S_ISDIR(mode) and not stat.S_ISREG(mode):
			raise LauncherUpdateError(f"launcher release contains a special file: {item}")
	metadata = _read_release_metadata(release)
	version = metadata.get("version")
	if not isinstance(version, str):
		raise LauncherUpdateError("launcher release metadata has no version")
	parsed_version = _parse_version(version, "launcher release version")
	channel = metadata.get("channel")
	if channel is not None and channel != parsed_version.channel:
		raise LauncherUpdateError("launcher release channel does not match its version")
	tag = metadata.get("tag")
	if tag is not None and tag != parsed_version.tag:
		raise LauncherUpdateError("launcher release tag does not match its version")
	for required in (
		Path("ptlinux/__init__.py"),
		Path("ptlinux/__main__.py"),
		Path("ptlinux/cli.py"),
		Path("ptlinux/self_update.py"),
		Path("ptlinux") / PUBLIC_KEY_FILENAME,
		Path("ptdiag/__init__.py"),
		Path("ptdiag/assbin.py"),
		Path("ptdiag/logs.py"),
	):
		if not (release / required).is_file():
			raise LauncherUpdateError(f"launcher release is missing {required}")
	bundled_gamescope = (
		release
		/ "ptlinux/native/gamescope-3.16.20-ubuntu26.04-amd64"
	)
	if bundled_gamescope.exists():
		mode = bundled_gamescope.stat().st_mode
		if not stat.S_ISREG(mode) or not mode & 0o111:
			raise LauncherUpdateError(
				"launcher release contains a non-executable Gamescope compatibility binary"
			)
	return version


def _self_check_release(release: Path, expected_version: str) -> None:
	check = (
		"import sys; "
		"sys.path.insert(0, sys.argv[1]); "
		"import ptlinux, ptlinux.cli, ptlinux.self_update, ptlinux.ui; "
		"raise SystemExit(0 if ptlinux.__version__ == sys.argv[2] else 1)"
	)
	try:
		result = subprocess.run(
			[sys.executable, "-I", "-B", "-c", check, str(release), expected_version],
			capture_output=True,
			text=True,
			encoding="utf-8",
			errors="replace",
			timeout=15,
			check=False,
		)
	except (OSError, subprocess.TimeoutExpired) as exc:
		raise LauncherUpdateError(f"could not test the staged launcher: {exc}") from exc
	if result.returncode != 0:
		detail = (result.stderr or result.stdout).strip()
		raise LauncherUpdateError(
			"staged launcher self-check failed" + (f": {detail}" if detail else "")
		)


@contextmanager
def _launcher_mutation_lock(paths: AppPaths, operation: str) -> Iterator[None]:
	"""Exclude game launches and concurrent release-link mutations."""

	with launcher_operation_lock(
		paths.launcher_operation_lock,
		operation,
		active_game_probe=lambda: managed_game_process_is_active(
			paths.runtime_view,
			(paths.proton_prefix, paths.umu_prefix),
		),
	):
		with launcher_operation_lock(paths.launcher_update_lock, operation):
			yield


def _precheck_release_install(paths: AppPaths, manifest: UpdateManifest) -> None:
	"""Reject a release before download when local transaction prerequisites fail."""

	_validate_rollback_config_compatibility(paths, manifest)
	for directory, label in (
		(paths.downloads, "launcher download cache"),
		(paths.app_home, "launcher application directory"),
	):
		directory.mkdir(parents=True, exist_ok=True)
		if directory.is_symlink() or not directory.is_dir() or not os.access(directory, os.W_OK):
			raise LauncherUpdateError(f"{label} is not a writable directory: {directory}")
	archive = paths.launcher_update_archive(manifest.version)
	cache_bytes = 0 if _archive_matches(archive, manifest) else manifest.archive_size
	extracted_bytes = (
		sum(file.size for file in manifest.files)
		if manifest.files
		else min(MAX_EXTRACTED_BYTES, manifest.archive_size * 4)
	)
	_require_free_space(paths.downloads, cache_bytes + 1024 * 1024, "download")
	_require_free_space(paths.app_home, extracted_bytes + 1024 * 1024, "staging")


def _require_free_space(path: Path, required_bytes: int, label: str) -> None:
	try:
		free_bytes = shutil.disk_usage(path).free
	except OSError as exc:
		raise LauncherUpdateError(f"cannot inspect free space for {path}: {exc}") from exc
	if free_bytes < required_bytes:
		raise LauncherUpdateError(
			f"insufficient free disk space for launcher {label}: "
			f"need {required_bytes} bytes, have {free_bytes}"
		)


def _current_config_schema(paths: AppPaths) -> int:
	from .settings import SETTINGS_SCHEMA

	try:
		data = paths.settings_file.read_bytes()
	except FileNotFoundError:
		return SETTINGS_SCHEMA
	except OSError as exc:
		raise LauncherUpdateError(f"cannot inspect launcher settings schema: {exc}") from exc
	if len(data) > 1024 * 1024:
		raise LauncherUpdateError("launcher settings file is too large to validate")
	try:
		value = json.loads(data.decode("utf-8-sig"), object_pairs_hook=_unique_json_object)
	except (UnicodeError, json.JSONDecodeError, ValueError) as exc:
		raise LauncherUpdateError(f"launcher settings are invalid: {exc}") from exc
	if not isinstance(value, dict):
		raise LauncherUpdateError("launcher settings must contain an object")
	schema = value.get("schema", SETTINGS_SCHEMA)
	if isinstance(schema, bool) or not isinstance(schema, int) or schema <= 0:
		raise LauncherUpdateError("launcher settings schema is invalid")
	return schema


def _validate_rollback_config_compatibility(
	paths: AppPaths,
	manifest: UpdateManifest,
) -> None:
	_validate_config_schema_range(
		_current_config_schema(paths),
		manifest.minimum_config_schema,
		manifest.config_schema,
		f"launcher {manifest.version}",
	)


def _validate_rollback_release_compatibility(paths: AppPaths, release: Path) -> None:
	metadata = _read_release_metadata(release)
	config_schema = metadata.get("config_schema", 1)
	minimum_config_schema = metadata.get("minimum_config_schema", 1)
	if (
		isinstance(config_schema, bool)
		or not isinstance(config_schema, int)
		or isinstance(minimum_config_schema, bool)
		or not isinstance(minimum_config_schema, int)
	):
		raise LauncherUpdateError("launcher release has invalid config compatibility")
	_validate_config_schema_range(
		_current_config_schema(paths),
		minimum_config_schema,
		config_schema,
		f"launcher {metadata.get('version', release.name)}",
	)


def _validate_config_schema_range(
	current: int,
	minimum: int,
	maximum: int,
	label: str,
) -> None:
	if minimum <= 0 or maximum <= 0 or minimum > maximum:
		raise LauncherUpdateError(f"{label} has invalid config schema compatibility")
	if current < minimum or current > maximum:
		raise LauncherUpdateError(
			f"{label} supports config schemas {minimum} through {maximum}, "
			f"but the installed configuration uses schema {current}"
		)


def _read_release_metadata(release: Path) -> dict[str, object]:
	metadata_path = release / "release.json"
	try:
		metadata = json.loads(
			metadata_path.read_text(encoding="utf-8"),
			object_pairs_hook=_unique_json_object,
		)
	except (OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc:
		raise LauncherUpdateError(f"launcher release metadata is invalid: {exc}") from exc
	if not isinstance(metadata, dict) or metadata.get("schema") != 1:
		raise LauncherUpdateError("launcher release metadata has an unsupported schema")
	return metadata


def _activate_release(
	paths: AppPaths,
	release: Path,
	*,
	telemetry_metadata: Mapping[str, object] | None = None,
	restore_reason: str = "failed_update",
) -> Path | None:
	if paths.app_home.is_symlink():
		raise LauncherUpdateError(f"launcher application home cannot be a link: {paths.app_home}")
	paths.app_home.mkdir(parents=True, exist_ok=True, mode=0o755)
	paths.app_home.chmod(0o755)
	_recover_interrupted_activation(paths)
	current = _release_from_link(paths.launcher_current, paths.launcher_releases)
	previous = _release_from_link(paths.launcher_previous, paths.launcher_releases)
	if paths.launcher_current.exists() and not paths.launcher_current.is_symlink():
		raise LauncherUpdateError(
			f"launcher current path is not a symbolic link: {paths.launcher_current}"
		)
	if paths.launcher_previous.exists() and not paths.launcher_previous.is_symlink():
		raise LauncherUpdateError(
			f"launcher previous path is not a symbolic link: {paths.launcher_previous}"
		)
	resolved_release = release.resolve()
	_validate_release_directory(resolved_release)
	if current == resolved_release:
		return current
	_write_activation_journal(
		paths,
		state="applying",
		current=current,
		previous=previous,
		target=resolved_release,
	)
	try:
		if current is not None:
			_validate_release_directory(current)
			_replace_release_link(
				paths.launcher_previous,
				current,
				paths.launcher_releases,
			)
		_replace_release_link(
			paths.launcher_current,
			resolved_release,
			paths.launcher_releases,
		)
		_fsync_directory(paths.app_home)
		activated = _release_from_link(paths.launcher_current, paths.launcher_releases)
		retained_previous = _release_from_link(
			paths.launcher_previous,
			paths.launcher_releases,
		)
		expected_previous = current if current is not None else previous
		if activated is None or activated != resolved_release or retained_previous != expected_previous:
			raise LauncherUpdateError(
				"launcher release links failed their post-activation health check"
			)
		assert activated is not None
		_validate_release_directory(activated)
		_write_activation_journal(
			paths,
			state="activated",
			current=current,
			previous=previous,
			target=resolved_release,
		)
	except BaseException as activation_error:
		restore_metadata = {
			**dict(telemetry_metadata or {}),
			"reason": restore_reason,
		}
		_send_launcher_release_event(
			"rollback_restore_started",
			metadata=restore_metadata,
			operation="launcher_rollback",
			stage="restore",
		)
		try:
			_restore_release_links(paths, current=current, previous=previous)
		except BaseException as restore_error:
			_send_launcher_release_event(
				"rollback_restore_failed",
				metadata=restore_metadata,
				operation="launcher_rollback",
				stage="restore",
				severity="critical",
				message=str(restore_error),
			)
			raise LauncherUpdateError(
				f"launcher activation failed ({activation_error}); "
				f"snapshot restoration also failed ({restore_error})"
			) from activation_error
		paths.launcher_update_transaction.unlink(missing_ok=True)
		_fsync_directory(paths.launcher_update_transaction.parent)
		_send_launcher_release_event(
			"rollback_restore_success",
			metadata=restore_metadata,
			operation="launcher_rollback",
			stage="restore",
			severity="warning",
		)
		raise
	paths.launcher_update_transaction.unlink(missing_ok=True)
	_fsync_directory(paths.launcher_update_transaction.parent)
	return current


def _recover_interrupted_activation(paths: AppPaths) -> None:
	journal = paths.launcher_update_transaction
	try:
		if journal.is_symlink():
			raise LauncherUpdateError(f"launcher transaction journal cannot be a link: {journal}")
		value = json.loads(
			journal.read_text(encoding="utf-8"),
			object_pairs_hook=_unique_json_object,
		)
	except FileNotFoundError:
		return
	except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
		raise LauncherUpdateError(f"launcher transaction journal is invalid: {exc}") from exc
	if not isinstance(value, dict) or value.get("schema") != 1:
		raise LauncherUpdateError("launcher transaction journal has an unsupported schema")
	state = value.get("state")
	if state not in {"applying", "activated"}:
		raise LauncherUpdateError("launcher transaction journal has an invalid state")
	current = _journal_release(paths, value.get("current"), "current")
	previous = _journal_release(paths, value.get("previous"), "previous")
	target_version = _journal_version(value.get("target"), "target")
	target_path = paths.launcher_releases / target_version
	try:
		_validate_release_directory(target_path)
	except LauncherUpdateError:
		target = None
	else:
		target = target_path.resolve()
	if state == "activated":
		active = _release_from_link(paths.launcher_current, paths.launcher_releases)
		retained_previous = _release_from_link(
			paths.launcher_previous,
			paths.launcher_releases,
		)
		expected_previous = current if current is not None else previous
		if target is not None and active == target and retained_previous == expected_previous:
			journal.unlink()
			_fsync_directory(journal.parent)
			return
	metadata = _recovery_telemetry_metadata(current, target_version)
	_send_launcher_release_event(
		"rollback_restore_started",
		metadata=metadata,
		operation="launcher_rollback",
		stage="restore",
		severity="warning",
	)
	try:
		_restore_release_links(paths, current=current, previous=previous)
	except BaseException as exc:
		_send_launcher_release_event(
			"rollback_restore_failed",
			metadata=metadata,
			operation="launcher_rollback",
			stage="restore",
			severity="critical",
			message=str(exc),
		)
		raise
	journal.unlink()
	_fsync_directory(journal.parent)
	_send_launcher_release_event(
		"rollback_restore_success",
		metadata=metadata,
		operation="launcher_rollback",
		stage="restore",
		severity="warning",
	)


def _recovery_telemetry_metadata(
	current: Path | None,
	target_version: str,
) -> dict[str, object]:
	current_version = current.name if current is not None else __version__
	try:
		current_channel = SemanticVersion.parse(current_version).channel
	except ReleaseModelError:
		current_channel = STABLE_CHANNEL
	target_channel = SemanticVersion.parse(target_version).channel
	return _launcher_release_metadata(
		current_version=current_version,
		target_version=target_version,
		current_channel=current_channel,
		target_channel=target_channel,
		selected_channel=target_channel,
		tag=SemanticVersion.parse(target_version).tag,
		manifest_schema=None,
		manifest_sha256=None,
		rollback=True,
		reason="failed_update",
	)


def _journal_version(value: object, label: str) -> str:
	if not isinstance(value, str):
		raise LauncherUpdateError(f"launcher transaction {label} release is invalid")
	try:
		SemanticVersion.parse(value)
	except ReleaseModelError as exc:
		raise LauncherUpdateError(
			f"launcher transaction {label} version is invalid"
		) from exc
	return value


def _journal_release(paths: AppPaths, value: object, label: str) -> Path | None:
	if value is None:
		return None
	version = _journal_version(value, label)
	release = paths.launcher_releases / version
	_validate_release_directory(release)
	return release.resolve()


def _restore_release_links(
	paths: AppPaths,
	*,
	current: Path | None,
	previous: Path | None,
) -> None:
	_set_release_link(paths.launcher_current, current, paths.launcher_releases)
	_set_release_link(paths.launcher_previous, previous, paths.launcher_releases)
	_fsync_directory(paths.app_home)


def _set_release_link(link: Path, release: Path | None, releases: Path) -> None:
	if release is not None:
		_replace_release_link(link, release, releases)
		return
	if link.is_symlink():
		link.unlink()
	elif link.exists():
		raise LauncherUpdateError(f"launcher release link is not symbolic: {link}")


def _write_activation_journal(
	paths: AppPaths,
	*,
	state: str,
	current: Path | None,
	previous: Path | None,
	target: Path,
) -> None:
	journal = paths.launcher_update_transaction
	payload = {
		"schema": 1,
		"state": state,
		"current": current.name if current is not None else None,
		"previous": previous.name if previous is not None else None,
		"target": target.name,
		"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
	}
	journal.parent.mkdir(parents=True, exist_ok=True)
	temporary = journal.with_name(f".{journal.name}.{os.getpid()}.tmp")
	try:
		with temporary.open("w", encoding="utf-8", newline="\n") as handle:
			json.dump(payload, handle, indent=2, sort_keys=True)
			handle.write("\n")
			handle.flush()
			os.fsync(handle.fileno())
		temporary.chmod(0o600)
		os.replace(temporary, journal)
		_fsync_directory(journal.parent)
	except BaseException:
		temporary.unlink(missing_ok=True)
		raise


def _replace_release_link(link: Path, release: Path, releases: Path) -> None:
	resolved_release = release.resolve()
	resolved_releases = releases.resolve()
	if resolved_release.parent != resolved_releases:
		raise LauncherUpdateError(f"release is outside the launcher release directory: {release}")
	_validate_release_directory(resolved_release)
	temporary = link.with_name(f".{link.name}.{os.getpid()}.tmp")
	temporary.unlink(missing_ok=True)
	try:
		temporary.symlink_to(Path("releases") / resolved_release.name)
		os.replace(temporary, link)
	except BaseException:
		temporary.unlink(missing_ok=True)
		raise


def _release_from_link(link: Path, releases: Path) -> Path | None:
	if not link.is_symlink():
		if link.exists():
			raise LauncherUpdateError(f"launcher release link is not symbolic: {link}")
		return None
	try:
		target_text = os.readlink(link)
	except OSError as exc:
		raise LauncherUpdateError(f"cannot read launcher release link {link}: {exc}") from exc
	target = (link.parent / target_text).resolve()
	root = releases.resolve()
	if target.parent != root:
		raise LauncherUpdateError(f"launcher release link escapes its release directory: {link}")
	if not target.is_dir() or target.is_symlink():
		raise LauncherUpdateError(f"launcher release link is broken: {link}")
	return target


def _archive_matches(path: Path, manifest: UpdateManifest) -> bool:
	try:
		return (
			path.stat().st_size == manifest.archive_size
			and _sha256_file(path) == manifest.archive_sha256
		)
	except OSError:
		return False


def _tree_digest(root: Path) -> str:
	digest = hashlib.sha256()
	root_mode = root.stat().st_mode
	digest.update(b"R\x00" + str(root_mode & 0o777).encode("ascii") + b"\x00")
	for item in sorted(root.rglob("*"), key=lambda value: value.relative_to(root).as_posix()):
		relative = item.relative_to(root).as_posix().encode("utf-8")
		if item.is_symlink():
			raise LauncherUpdateError(f"launcher release contains a symbolic link: {item}")
		state = item.stat()
		mode = str(state.st_mode & 0o777).encode("ascii")
		if stat.S_ISDIR(state.st_mode):
			digest.update(b"D\x00" + relative + b"\x00" + mode + b"\x00")
		elif stat.S_ISREG(state.st_mode):
			digest.update(b"F\x00" + relative + b"\x00")
			digest.update(mode + b"\x00")
			digest.update(str(state.st_size).encode("ascii") + b"\x00")
			with item.open("rb") as handle:
				for chunk in iter(lambda: handle.read(READ_SIZE), b""):
					digest.update(chunk)
			digest.update(b"\x00")
		else:
			raise LauncherUpdateError(f"launcher release contains a special file: {item}")
	return digest.hexdigest()


def _sha256_file(path: Path) -> str:
	digest = hashlib.sha256()
	with path.open("rb") as handle:
		for chunk in iter(lambda: handle.read(READ_SIZE), b""):
			digest.update(chunk)
	return digest.hexdigest()


def _signature_url(manifest_url: str) -> str:
	parts = urlsplit(manifest_url)
	return urlunsplit((parts.scheme, parts.netloc, parts.path + ".sig", parts.query, ""))


def _resource_url(value: str) -> str:
	parts = urlsplit(value)
	return urlunsplit(
		(parts.scheme.casefold(), parts.netloc.casefold(), parts.path, parts.query, "")
	)


def _require_https_url(value: str, label: str) -> None:
	parts = urlsplit(value)
	if (
		parts.scheme.casefold() != "https"
		or not parts.netloc
		or parts.username is not None
		or parts.password is not None
		or parts.fragment
	):
		raise LauncherUpdateError(f"{label} must be an HTTPS URL without credentials")


def _parse_version(value: str, label: str) -> SemanticVersion:
	try:
		return SemanticVersion.parse(value)
	except ReleaseModelError as exc:
		raise LauncherUpdateError(f"invalid {label}: {value!r}: {exc}") from exc


def _parse_published_at(value: str) -> datetime:
	try:
		parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
	except ValueError as exc:
		raise LauncherUpdateError(
			f"manifest published_at must be an ISO-8601 timestamp: {value!r}"
		) from exc
	if parsed.tzinfo is None or parsed.utcoffset() is None:
		raise LauncherUpdateError("manifest published_at must include a timezone")
	return parsed


def _cached_timestamp(value: object) -> str | None:
	if not isinstance(value, str) or not value:
		return None
	try:
		parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
	except ValueError:
		return None
	if parsed.tzinfo is None or parsed.utcoffset() is None:
		return None
	return value


def _release_version(
	releases: list[LauncherInstalledRelease],
	*,
	active: bool = False,
	previous: bool = False,
) -> str | None:
	for release in releases:
		if (active and release.active) or (previous and release.previous):
			return release.version
	return None


def _required_string(value: dict[str, object], key: str) -> str:
	item = value.get(key)
	if not isinstance(item, str) or not item:
		raise LauncherUpdateError(f"update manifest field {key!r} must be a string")
	return item


def _required_integer(value: dict[str, object], key: str) -> int:
	item = value.get(key)
	if isinstance(item, bool) or not isinstance(item, int):
		raise LauncherUpdateError(f"update manifest field {key!r} must be an integer")
	return item


def _required_positive_integer(value: dict[str, object], key: str) -> int:
	item = _required_integer(value, key)
	if item <= 0:
		raise LauncherUpdateError(
			f"update manifest field {key!r} must be a positive integer"
		)
	return item


def _optional_positive_integer(value: dict[str, object], key: str) -> int | None:
	item = value.get(key)
	if item is None:
		return None
	if isinstance(item, bool) or not isinstance(item, int) or item <= 0:
		raise LauncherUpdateError(
			f"release history field {key!r} must be a positive integer or null"
		)
	return item


def _unique_json_object(pairs: list[tuple[str, object]]) -> dict[str, object]:
	result: dict[str, object] = {}
	for key, value in pairs:
		if key in result:
			raise ValueError(f"duplicate JSON key: {key}")
		result[key] = value
	return result


def _content_length(value: str | None, label: str) -> int | None:
	if value is None:
		return None
	try:
		length = int(value)
	except ValueError as exc:
		raise LauncherUpdateError(f"{label} returned an invalid Content-Length") from exc
	if length < 0:
		raise LauncherUpdateError(f"{label} returned an invalid Content-Length")
	return length


def _write_json(path: Path, value: object) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
	try:
		with temporary.open("w", encoding="utf-8") as handle:
			json.dump(value, handle, indent=2, sort_keys=True)
			handle.write("\n")
			handle.flush()
			os.fsync(handle.fileno())
		temporary.replace(path)
		_fsync_directory(path.parent)
	except BaseException:
		temporary.unlink(missing_ok=True)
		raise


def _record_launcher_operation(
	paths: AppPaths,
	*,
	from_version: str,
	to_version: str,
	from_channel: str,
	to_channel: str,
	selected_channel: str,
	result: str,
	rollback: bool,
	reason: str,
	tag: str | None = None,
	manifest_schema: int | None = None,
	manifest_sha256: str | None = None,
) -> bool:
	"""Append one bounded diagnostic record without affecting an installation."""

	path = paths.launcher_update_operations
	try:
		try:
			value = json.loads(path.read_text(encoding="utf-8"))
		except FileNotFoundError:
			value = {"schema": LOCAL_OPERATION_HISTORY_SCHEMA, "operations": []}
		if not isinstance(value, dict) or value.get("schema") != LOCAL_OPERATION_HISTORY_SCHEMA:
			return False
		operations = value.get("operations")
		if not isinstance(operations, list):
			return False
		operations = [item for item in operations if isinstance(item, dict)]
		operations.append(
			{
				"at": datetime.now().astimezone().isoformat(timespec="seconds"),
				"from": from_version,
				"from_channel": from_channel,
				"manifest_hash": manifest_sha256,
				"manifest_version": manifest_schema,
				"reason": reason,
				"result": result,
				"rollback": rollback,
				"selected_channel": selected_channel,
				"tag": tag,
				"to": to_version,
				"to_channel": to_channel,
			}
		)
		_write_json(
			path,
			{
				"schema": LOCAL_OPERATION_HISTORY_SCHEMA,
				"operations": operations[-MAX_LOCAL_OPERATION_HISTORY:],
			},
		)
	except (OSError, UnicodeError, json.JSONDecodeError, ValueError, RuntimeError):
		return False
	return True


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 update_check_to_dict(check: LauncherUpdateCheck) -> dict[str, object]:
	"""Return an explicitly public JSON representation for CLI/UI callers."""

	return asdict(check)