"""Diagnose and narrowly repair the Bubblewrap host sandbox used by UMU."""

from __future__ import annotations

from dataclasses import asdict, dataclass, replace
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
import time
from typing import Callable, Mapping, Sequence

from . import __version__
from .paths import AppPaths


UID_MAP_PERMISSION_DENIED = "bwrap: setting up uid map: permission denied"
_BWRAP_ATTACHMENT_PATTERN = re.compile(
	r"(?m)^[ \t]*(?:profile[ \t]+[^\s{]+[ \t]+)?/usr/bin/bwrap(?=[ \t{])"
)
_MAX_APPARMOR_PROFILE_BYTES = 1024 * 1024

# AppArmor upstream's restrictive Bubblewrap profile, pinned to a commit and
# its AppArmor 4.0 ABI so it works on Ubuntu 24.04. It grants Bubblewrap the
# setup capabilities it needs, then stacks children under unpriv_bwrap, where
# capabilities are explicitly denied. This is intentionally not an unconfined
# profile and does not disable Ubuntu's global user-namespace restriction.
BWRAP_APPARMOR_PROFILE = """# Installed by the PristonTale EU Linux launcher.
# Upstream source:
# https://gitlab.com/apparmor/apparmor/-/blob/8e431ebcd915216a03ebc8d01e72b1741bb2f855/profiles/apparmor/profiles/extras/bwrap-userns-restrict
#
# This profile allows almost everything and only exists to allow bwrap
# to work on a system with user namespace restrictions being enforced.
# bwrap is allowed access to user namespaces and capabilities within
# the user namespace, but its children do not have capabilities,
# blocking bwrap from being able to be used to arbitrarily by-pass the
# user namespace restrictions.

# Note: the bwrap child is stacked against the bwrap profile due to
# bwraps use of no-new-privs.

abi <abi/4.0>,

include <tunables/global>

profile bwrap /usr/bin/bwrap flags=(attach_disconnected,mediate_deleted) {
  allow capability,
  # not allow all, to allow for pix stack on systems that don't support
  # rule priority.
  #
  # sadly we have to allow 'm' every where to allow children to work under
  # profile stacking atm.
  allow file rwlkm /{**,},
  allow network,
  allow unix,
  allow ptrace,
  allow signal,
  allow mqueue,
  allow io_uring,
  allow userns,
  allow mount,
  allow umount,
  allow pivot_root,
  allow dbus,

  # stacked like this due to no-new-privs restriction
  # this will stack a target profile against bwrap and unpriv_bwrap
  # Ideally
  # - there would be a transition at userns creation first. This would allow
  #   for the bwrap profile to be tighter, and looser within the user
  #   ns. bwrap will still have to fairly loose until a transition at
  #   namespacing in general (not just user ns) is available.
  # - there would be an independent second target as fallback
  #   This would allow for select target profiles to be used, and not
  #   necessarily stack the unpriv_bwrap in cases where this is desired
  #
  # the ix works here because stack will apply to ix fallback
  # Ideally we would sanitize the environment across a privilege boundry
  # (leaving bwarp into application) but flatpak etc use environment glibc
  # sanitized environment variables as part of the sandbox setup.
  allow pix /** -> &bwrap//&unpriv_bwrap,

  # the local include should not be used without understanding the userns
  # restriction.
  # Site-specific additions and overrides. See local/README for details.
  include if exists <local/bwrap-userns-restrict>
}

# The unpriv_bwrap profile is used to strip capabilities within the userns
profile unpriv_bwrap flags=(attach_disconnected,mediate_deleted) {
  # not allow all, to allow for pix stack
  allow file rwlkm /{**,},
  allow network,
  allow unix,
  allow ptrace,
  allow signal,
  allow mqueue,
  allow io_uring,
  allow userns,
  allow mount,
  allow umount,
  allow pivot_root,
  allow dbus,

  # bwrap profile does stacking against itself this will keep the target
  # profile from having elevated privileges in the container.
  # If done recursively the stack will remove any duplicate
  allow pix /** -> &unpriv_bwrap,

  audit deny capability,

  # the local include should not be used without understanding the userns
  # restriction.
  # Site-specific additions and overrides. See local/README for details.
  include if exists <local/unpriv_bwrap>
}
"""
BWRAP_APPARMOR_PROFILE_SHA256 = hashlib.sha256(
	BWRAP_APPARMOR_PROFILE.encode("utf-8")
).hexdigest()


@dataclass(frozen=True)
class ContainerHostFiles:
	"""Fixed host paths, injectable so repair logic can be tested safely."""

	bwrap: Path = Path("/usr/bin/bwrap")
	probe_program: Path = Path("/usr/bin/true")
	os_release: Path = Path("/etc/os-release")
	apparmor_parser: Path = Path("/usr/sbin/apparmor_parser")
	apparmor_directory: Path = Path("/etc/apparmor.d")
	install: Path = Path("/usr/bin/install")
	link: Path = Path("/usr/bin/ln")
	remove: Path = Path("/usr/bin/rm")
	obsolete_umu_profile_target: Path = Path(
		"/usr/share/apparmor/extra-profiles/bwrap-userns-restrict"
	)
	apparmor_restrict_userns: Path = Path(
		"/proc/sys/kernel/apparmor_restrict_unprivileged_userns"
	)
	unprivileged_userns_clone: Path = Path(
		"/proc/sys/kernel/unprivileged_userns_clone"
	)
	max_user_namespaces: Path = Path("/proc/sys/user/max_user_namespaces")

	@property
	def distribution_profile(self) -> Path:
		return self.apparmor_directory / "bwrap-userns-restrict"

	@property
	def managed_profile(self) -> Path:
		return self.apparmor_directory / "bwrap-userns-restrict-ptlinux"

	@property
	def umu_profile_link(self) -> Path:
		return self.apparmor_directory / "bwrap-userns-restrict-umu"


DEFAULT_CONTAINER_HOST_FILES = ContainerHostFiles()
CommandExecutor = Callable[[Sequence[str], float], subprocess.CompletedProcess[str]]


@dataclass(frozen=True)
class ContainerHostReport:
	ready: bool
	issue: str | None
	distribution: str
	bwrap_path: str | None
	probe_returncode: int | None
	probe_stderr: str
	apparmor_restrict_unprivileged_userns: str | None
	unprivileged_userns_clone: str | None
	max_user_namespaces: str | None
	active_profile_source: str | None
	conflicting_bwrap_profile: str | None
	broken_umu_profile_link: bool
	repairable: bool
	recommendation: str | None
	repair_attempted: bool = False
	repaired: bool = False
	repair_actions: tuple[str, ...] = ()
	repair_error: str | None = None

	def to_dict(self) -> dict[str, object]:
		return asdict(self)


def inspect_container_host(
	*,
	environment: Mapping[str, str] | None = None,
	files: ContainerHostFiles = DEFAULT_CONTAINER_HOST_FILES,
	command_executor: CommandExecutor | None = None,
	timeout_seconds: float = 5.0,
) -> ContainerHostReport:
	"""Run a harmless user-namespace probe and classify the exact failure."""

	_ = environment
	distribution_data = _read_os_release(files.os_release)
	distribution = distribution_data.get("ID", "unknown")
	bwrap_available = files.bwrap.is_file() and os.access(files.bwrap, os.X_OK)
	restrict_userns = _read_setting(files.apparmor_restrict_userns)
	userns_clone = _read_setting(files.unprivileged_userns_clone)
	max_user_namespaces = _read_setting(files.max_user_namespaces)
	profile = _active_profile_source(files)
	broken_link = _known_broken_umu_profile_link(files)
	returncode: int | None = None
	stderr = ""
	issue: str | None = None
	if not bwrap_available:
		issue = "bubblewrap-missing"
	else:
		command = (
			str(files.bwrap),
			"--ro-bind",
			"/",
			"/",
			"--unshare-user",
			"--",
			str(files.probe_program),
		)
		try:
			result = _run(command, timeout_seconds, command_executor)
			returncode = result.returncode
			stderr = _bounded_output(result.stderr)
		except subprocess.TimeoutExpired as exc:
			issue = "bubblewrap-probe-timeout"
			stderr = _bounded_output(_timeout_stderr(exc))
		except OSError as exc:
			issue = "bubblewrap-probe-error"
			stderr = str(exc)
		if issue is None and returncode != 0:
			folded = stderr.casefold()
			if UID_MAP_PERMISSION_DENIED in folded:
				issue = "uid-map-permission-denied"
			elif userns_clone == "0":
				issue = "unprivileged-userns-disabled"
			elif max_user_namespaces == "0":
				issue = "user-namespace-limit-zero"
			else:
				issue = "bubblewrap-probe-failed"

	conflicting_profile = (
		_find_conflicting_bwrap_profile(files)
		if issue == "uid-map-permission-denied" or broken_link
		else None
	)
	ubuntu_family = _is_ubuntu_family(distribution_data)
	managed_collision = _managed_profile_collision(files)
	repairable = (
		ubuntu_family
		and restrict_userns == "1"
		and files.apparmor_parser.is_file()
		and os.access(files.apparmor_parser, os.X_OK)
		and bwrap_available
		and not managed_collision
		and conflicting_profile is None
		and (issue == "uid-map-permission-denied" or broken_link)
	)
	recommendation = _recommendation(
		issue,
		repairable=repairable,
		broken_link=broken_link,
		conflicting_profile=conflicting_profile,
		stderr=stderr,
	)
	return ContainerHostReport(
		ready=issue is None,
		issue=issue,
		distribution=distribution,
		bwrap_path=str(files.bwrap) if bwrap_available else None,
		probe_returncode=returncode,
		probe_stderr=stderr,
		apparmor_restrict_unprivileged_userns=restrict_userns,
		unprivileged_userns_clone=userns_clone,
		max_user_namespaces=max_user_namespaces,
		active_profile_source=str(profile) if profile else None,
		conflicting_bwrap_profile=(
			str(conflicting_profile) if conflicting_profile else None
		),
		broken_umu_profile_link=broken_link,
		repairable=repairable,
		recommendation=recommendation,
	)


def repair_container_host(
	*,
	environment: Mapping[str, str] | None = None,
	files: ContainerHostFiles = DEFAULT_CONTAINER_HOST_FILES,
	command_executor: CommandExecutor | None = None,
	privileged_executor: CommandExecutor | None = None,
) -> ContainerHostReport:
	"""Repair only known Ubuntu AppArmor states, then repeat the probe."""

	before = inspect_container_host(
		environment=environment,
		files=files,
		command_executor=command_executor,
	)
	started_at = time.monotonic()
	_send_dependency_fix_event(
		"dependency_fix_requested",
		metadata={"reason": "container_host", "issue": before.issue},
		stage="requested",
	)
	_send_dependency_fix_event(
		"dependency_fix_started",
		metadata={"issue": before.issue, "repairable": before.repairable},
		stage="started",
	)
	if before.ready and not before.broken_umu_profile_link:
		_send_dependency_fix_outcome(before, started_at)
		return before
	if not before.repairable:
		result = replace(
			before,
			repair_attempted=True,
			repair_error=(
				before.recommendation or "No safe automatic repair is available."
			),
		)
		_send_dependency_fix_outcome(result, started_at)
		return result
	actions: list[str] = []
	created_managed_profile = False
	profile_loaded = False
	selected_profile = _active_profile_source(files)
	try:
		if selected_profile is None:
			selected_profile = files.managed_profile
			_validate_and_install_managed_profile(
				files,
				command_executor=command_executor,
				privileged_executor=privileged_executor,
			)
			created_managed_profile = True
			actions.append(f"installed {selected_profile}")
		else:
			_validate_profile(
				selected_profile,
				files,
				command_executor=command_executor,
			)

		loaded = _run_privileged(
			(str(files.apparmor_parser), "-r", str(selected_profile)),
			120.0,
			privileged_executor,
		)
		_require_success(loaded, f"could not load AppArmor profile {selected_profile}")
		profile_loaded = True
		actions.append(f"loaded {selected_profile}")

		if _known_broken_umu_profile_link(files):
			linked = _run_privileged(
				(
					str(files.link),
					"-sfn",
					"--",
					str(selected_profile),
					str(files.umu_profile_link),
				),
				120.0,
				privileged_executor,
			)
			_require_success(
				linked,
				"could not repair UMU's broken AppArmor profile link",
			)
			actions.append(f"repaired {files.umu_profile_link}")
	except (OSError, RuntimeError, subprocess.TimeoutExpired) as exc:
		rollback_error = None
		if created_managed_profile and not profile_loaded:
			rollback_error = _rollback_managed_profile(files, privileged_executor)
			if rollback_error is None:
				actions.append(f"rolled back {files.managed_profile}")
		after_failure = inspect_container_host(
			environment=environment,
			files=files,
			command_executor=command_executor,
		)
		error = str(exc)
		if rollback_error:
			error = f"{error}; rollback also failed: {rollback_error}"
		result = replace(
			after_failure,
			repair_attempted=True,
			repair_actions=tuple(actions),
			repair_error=error,
		)
		_send_dependency_fix_outcome(result, started_at)
		return result

	after = inspect_container_host(
		environment=environment,
		files=files,
		command_executor=command_executor,
	)
	error = None
	if not after.ready:
		error = (
			"The AppArmor profile loaded, but the Bubblewrap user-namespace probe "
			"still fails. "
			+ (after.recommendation or "Inspect the host security policy.")
		)
	result = replace(
		after,
		repair_attempted=True,
		repaired=after.ready,
		repair_actions=tuple(actions),
		repair_error=error,
	)
	_send_dependency_fix_outcome(result, started_at)
	return result


def _send_dependency_fix_outcome(
	report: ContainerHostReport,
	started_at: float,
) -> None:
	success = report.ready and not report.broken_umu_profile_link and (
		report.repair_error is None
	)
	_send_dependency_fix_event(
		"dependency_fix_success" if success else "dependency_fix_failed",
		severity="info" if success else "error",
		message=None if success else report.repair_error,
		metadata={
			"duration_ms": round((time.monotonic() - started_at) * 1000),
			"ready": report.ready,
			"repaired": report.repaired,
			"repair_attempted": report.repair_attempted,
			"action_count": len(report.repair_actions),
			"issue": report.issue,
		},
		stage="completed" if success else "repair",
	)


def _send_dependency_fix_event(
	event: str,
	*,
	severity: str = "info",
	message: str | None = None,
	metadata: Mapping[str, object] | None = None,
	stage: str,
) -> None:
	try:
		from .telemetry import current_telemetry

		telemetry = current_telemetry()
		if telemetry is not None:
			telemetry.send_event(
				event,
				severity=severity,
				message=message,
				metadata=metadata,
				context={"operation": "dependency_fix", "stage": stage},
			)
	except Exception:
		# Host recovery must remain independent from optional telemetry.
		pass


def repair_container_host_after_update(paths: AppPaths) -> ContainerHostReport:
	"""Run the repair once after activation, including upgrades by older code."""

	status = inspect_container_host()
	if status.ready and not status.broken_umu_profile_link:
		return status
	if not _launcher_was_updated_to_current_version(paths):
		return status
	marker = paths.container_host_repair_state
	if _repair_was_attempted(marker, status):
		return status
	print(
		"Launcher update: checking Steam Runtime sandbox permissions; "
		"authentication may be requested for a narrow AppArmor repair.",
		file=sys.stderr,
	)
	report = repair_container_host()
	try:
		_write_repair_marker(marker, report)
	except OSError as exc:
		print(
			f"Warning: could not record the container-host repair attempt: {exc}",
			file=sys.stderr,
		)
	return report


def container_host_error(report: ContainerHostReport) -> str:
	if report.ready and report.broken_umu_profile_link:
		return report.repair_error or report.recommendation or (
			"UMU's AppArmor profile link remains broken."
		)
	message = (
		"The Steam Runtime container cannot create its Bubblewrap user namespace."
	)
	if report.repair_error:
		return f"{message} Host repair failed: {report.repair_error}"
	if report.recommendation:
		return f"{message} {report.recommendation}"
	if report.probe_stderr:
		return f"{message} Bubblewrap reported: {report.probe_stderr}"
	return message


def _validate_and_install_managed_profile(
	files: ContainerHostFiles,
	*,
	command_executor: CommandExecutor | None,
	privileged_executor: CommandExecutor | None,
) -> None:
	if files.managed_profile.exists() or files.managed_profile.is_symlink():
		if _is_managed_profile(files.managed_profile):
			_validate_profile(
				files.managed_profile,
				files,
				command_executor=command_executor,
			)
			return
		raise RuntimeError(
			f"refusing to overwrite an unmanaged AppArmor file: {files.managed_profile}"
		)
	with tempfile.TemporaryDirectory(prefix="ptlinux-bwrap-profile-") as temporary:
		candidate = Path(temporary) / "bwrap-userns-restrict-ptlinux"
		candidate.write_text(BWRAP_APPARMOR_PROFILE, encoding="utf-8")
		candidate.chmod(0o600)
		_validate_profile(candidate, files, command_executor=command_executor)
		installed = _run_privileged(
			(
				str(files.install),
				"-m",
				"0644",
				"--",
				str(candidate),
				str(files.managed_profile),
			),
			120.0,
			privileged_executor,
		)
		_require_success(installed, "could not install the managed AppArmor profile")


def _validate_profile(
	profile: Path,
	files: ContainerHostFiles,
	*,
	command_executor: CommandExecutor | None,
) -> None:
	validated = _run(
		(str(files.apparmor_parser), "-Q", "-K", str(profile)),
		30.0,
		command_executor,
	)
	_require_success(validated, f"AppArmor rejected profile {profile}")


def _rollback_managed_profile(
	files: ContainerHostFiles,
	privileged_executor: CommandExecutor | None,
) -> str | None:
	try:
		removed = _run_privileged(
			(str(files.remove), "--", str(files.managed_profile)),
			120.0,
			privileged_executor,
		)
		_require_success(removed, "could not remove the newly installed profile")
	except (OSError, RuntimeError, subprocess.TimeoutExpired) as exc:
		return str(exc)
	return None


def _run(
	command: Sequence[str],
	timeout_seconds: float,
	executor: CommandExecutor | None,
) -> subprocess.CompletedProcess[str]:
	if executor is not None:
		return executor(tuple(command), timeout_seconds)
	return subprocess.run(
		list(command),
		capture_output=True,
		text=True,
		timeout=timeout_seconds,
		check=False,
	)


def _run_privileged(
	command: Sequence[str],
	timeout_seconds: float,
	executor: CommandExecutor | None,
) -> subprocess.CompletedProcess[str]:
	if executor is not None:
		return executor(tuple(command), timeout_seconds)
	prefix = _privilege_prefix()
	return subprocess.run(
		[*prefix, *command],
		capture_output=True,
		text=True,
		timeout=timeout_seconds,
		check=False,
	)


def _privilege_prefix() -> tuple[str, ...]:
	if os.geteuid() == 0:
		return ()
	if sys.stdin.isatty():
		for candidate in (Path("/usr/bin/sudo"), Path("/usr/bin/doas")):
			if candidate.is_file() and os.access(candidate, os.X_OK):
				return (str(candidate),)
	pkexec = Path("/usr/bin/pkexec")
	if pkexec.is_file() and os.access(pkexec, os.X_OK):
		return (str(pkexec),)
	sudo = Path("/usr/bin/sudo")
	if sudo.is_file() and os.access(sudo, os.X_OK):
		return str(sudo), "-n"
	raise RuntimeError(
		"root access is required for the AppArmor repair, but no usable "
		"sudo, doas, or pkexec command was found"
	)


def _require_success(result: subprocess.CompletedProcess[str], message: str) -> None:
	if result.returncode == 0:
		return
	detail = _bounded_output(result.stderr) or _bounded_output(result.stdout)
	if detail:
		raise RuntimeError(f"{message}: {detail}")
	raise RuntimeError(f"{message} (exit status {result.returncode})")


def _active_profile_source(files: ContainerHostFiles) -> Path | None:
	if files.distribution_profile.is_file():
		return files.distribution_profile
	if _trusted_umu_profile_link(files):
		return files.umu_profile_link
	if _is_managed_profile(files.managed_profile):
		return files.managed_profile
	return None


def _is_managed_profile(path: Path) -> bool:
	if not path.is_file() or path.is_symlink():
		return False
	try:
		content = path.read_bytes()
	except OSError:
		return False
	return hashlib.sha256(content).hexdigest() == BWRAP_APPARMOR_PROFILE_SHA256


def _managed_profile_collision(files: ContainerHostFiles) -> bool:
	path = files.managed_profile
	return (path.exists() or path.is_symlink()) and not _is_managed_profile(path)


def _find_conflicting_bwrap_profile(files: ContainerHostFiles) -> Path | None:
	known_names = {
		files.distribution_profile.name,
		files.managed_profile.name,
	}
	if _trusted_umu_profile_link(files):
		known_names.add(files.umu_profile_link.name)
	try:
		candidates = sorted(
			files.apparmor_directory.iterdir(),
			key=lambda path: path.name,
		)
	except OSError:
		return None
	for candidate in candidates:
		if candidate.name in known_names or not candidate.is_file():
			continue
		try:
			if candidate.stat().st_size > _MAX_APPARMOR_PROFILE_BYTES:
				continue
			content = candidate.read_text(encoding="utf-8")
		except (OSError, UnicodeError):
			continue
		policy = "\n".join(line.split("#", 1)[0] for line in content.splitlines())
		if _BWRAP_ATTACHMENT_PATTERN.search(policy):
			return candidate
	return None


def _trusted_umu_profile_link(files: ContainerHostFiles) -> bool:
	link = files.umu_profile_link
	if not link.is_symlink() or not link.is_file():
		return False
	try:
		target_text = os.readlink(link)
		target = Path(target_text)
		if not target.is_absolute():
			target = link.parent / target
		target = target.resolve()
		allowed = {
			files.distribution_profile.resolve(),
			files.managed_profile.resolve(),
			files.obsolete_umu_profile_target.resolve(),
		}
	except OSError:
		return False
	return target in allowed


def _known_broken_umu_profile_link(files: ContainerHostFiles) -> bool:
	link = files.umu_profile_link
	if not link.is_symlink():
		return False
	try:
		target_text = os.readlink(link)
	except OSError:
		return False
	if target_text != str(files.obsolete_umu_profile_target):
		return False
	return not files.obsolete_umu_profile_target.exists()


def _read_os_release(path: Path) -> dict[str, str]:
	try:
		lines = path.read_text(encoding="utf-8").splitlines()
	except OSError:
		return {}
	result: dict[str, str] = {}
	for line in lines:
		if "=" not in line or line.startswith("#"):
			continue
		key, value = line.split("=", 1)
		result[key] = value.strip().strip('"').strip("'")
	return result


def _is_ubuntu_family(os_release: Mapping[str, str]) -> bool:
	distribution = os_release.get("ID", "").casefold()
	like = set(os_release.get("ID_LIKE", "").casefold().split())
	return distribution == "ubuntu" or "ubuntu" in like


def _read_setting(path: Path) -> str | None:
	try:
		return path.read_text(encoding="ascii").strip()[:64]
	except (OSError, UnicodeError):
		return None


def _recommendation(
	issue: str | None,
	*,
	repairable: bool,
	broken_link: bool,
	conflicting_profile: Path | None,
	stderr: str,
) -> str | None:
	if issue is None:
		if broken_link:
			return (
				"UMU's AppArmor profile link is broken; run "
				"`pristontale-eu container-host --repair` before rebooting."
			)
		return None
	if issue == "bubblewrap-missing":
		return (
			"Install the distribution's bubblewrap package and run the check again."
		)
	if issue == "uid-map-permission-denied":
		if repairable:
			return (
				"Ubuntu AppArmor denied Bubblewrap's UID map. Run "
				"`pristontale-eu container-host --repair`; authentication is "
				"required."
			)
		if conflicting_profile is not None:
			return (
				"Bubblewrap's UID map was denied, but an unrelated AppArmor profile "
				f"already attaches to /usr/bin/bwrap: {conflicting_profile}. The "
				"launcher will not install a competing profile automatically."
			)
		return (
			"Bubblewrap's UID map was denied, but this host does not match the "
			"narrow Ubuntu AppArmor repair. Inspect the loaded bwrap policy."
		)
	if issue == "unprivileged-userns-disabled":
		return (
			"kernel.unprivileged_userns_clone is disabled. The launcher will not "
			"weaken this global security setting automatically."
		)
	if issue == "user-namespace-limit-zero":
		return (
			"user.max_user_namespaces is zero. The launcher will not change this "
			"global limit automatically."
		)
	if issue == "bubblewrap-probe-timeout":
		return (
			"The Bubblewrap sandbox probe timed out; inspect the host sandbox policy."
		)
	if stderr:
		return f"Bubblewrap's sandbox probe failed: {stderr}"
	return "Bubblewrap's sandbox probe failed without a diagnostic message."


def _timeout_stderr(error: subprocess.TimeoutExpired) -> str:
	value = error.stderr
	if isinstance(value, bytes):
		return value.decode("utf-8", errors="replace")
	return value or ""


def _bounded_output(value: str | None, limit: int = 2048) -> str:
	if not value:
		return ""
	cleaned = value.strip()
	if len(cleaned) <= limit:
		return cleaned
	return cleaned[:limit] + "…"


def _launcher_was_updated_to_current_version(paths: AppPaths) -> bool:
	try:
		data = paths.launcher_update_state.read_bytes()
		if len(data) > 64 * 1024:
			return False
		payload = json.loads(data)
	except (OSError, UnicodeError, json.JSONDecodeError):
		return False
	return isinstance(payload, dict) and payload.get("to_version") == __version__


def _repair_was_attempted(path: Path, status: ContainerHostReport) -> bool:
	try:
		data = path.read_bytes()
		if len(data) > 64 * 1024:
			return False
		payload = json.loads(data)
	except (OSError, UnicodeError, json.JSONDecodeError):
		return False
	return (
		isinstance(payload, dict)
		and payload.get("schema") == 1
		and payload.get("launcher_version") == __version__
		and payload.get("profile_sha256") == BWRAP_APPARMOR_PROFILE_SHA256
		and payload.get("issue") == status.issue
		and payload.get("broken_umu_profile_link") == status.broken_umu_profile_link
	)


def _write_repair_marker(path: Path, report: ContainerHostReport) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	payload = {
		"schema": 1,
		"attempted_at": datetime.now().astimezone().isoformat(),
		"launcher_version": __version__,
		"profile_sha256": BWRAP_APPARMOR_PROFILE_SHA256,
		"issue": report.issue,
		"broken_umu_profile_link": report.broken_umu_profile_link,
		"ready": report.ready,
		"repair_error": report.repair_error,
	}
	with tempfile.NamedTemporaryFile(
		mode="w",
		encoding="utf-8",
		dir=path.parent,
		prefix=f".{path.name}.",
		delete=False,
	) as handle:
		temporary = Path(handle.name)
		try:
			json.dump(payload, handle, indent=2, sort_keys=True)
			handle.write("\n")
			handle.flush()
			os.fsync(handle.fileno())
		except BaseException:
			temporary.unlink(missing_ok=True)
			raise
	try:
		temporary.chmod(0o600)
		temporary.replace(path)
	except BaseException:
		temporary.unlink(missing_ok=True)
		raise