Add channel-aware release tooling and updater recovery

2026-09-10 00:39:37 +02:00
parent cc48bf45d2
commit 0a89323533
42 changed files with 7840 additions and 0 deletions
--- a/ptlinux/self_update.py
+++ b/ptlinux/self_update.py
@@ -2253,6 +2253,7 @@ def _extract_release_archive(archive: Path, destination: Path, version: str) ->
 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}")
+	_remove_generated_bytecode_caches(release)
 	for item in (release, *release.rglob("*")):
 		if item.is_symlink():
 			raise LauncherUpdateError(f"launcher release contains a symbolic link: {item}")
@@ -2302,6 +2303,43 @@ def _validate_release_directory(release: Path) -> str:
 	return version


+def _remove_generated_bytecode_caches(release: Path) -> None:
+	"""Remove interpreter caches that are never part of a signed release."""
+
+	try:
+		cache_directories = sorted(
+			release.rglob("__pycache__"),
+			key=lambda path: len(path.parts),
+			reverse=True,
+		)
+	except OSError as exc:
+		raise LauncherUpdateError(
+			f"cannot inspect launcher bytecode caches in {release}: {exc}"
+		) from exc
+	for cache in cache_directories:
+		if cache.is_symlink() or not cache.is_dir():
+			raise LauncherUpdateError(f"launcher bytecode cache is unsafe: {cache}")
+		try:
+			entries = tuple(cache.iterdir())
+		except OSError as exc:
+			raise LauncherUpdateError(
+				f"cannot inspect launcher bytecode cache {cache}: {exc}"
+			) from exc
+		for entry in entries:
+			if entry.is_symlink() or not entry.is_file() or entry.suffix not in {".pyc", ".pyo"}:
+				raise LauncherUpdateError(
+					f"launcher bytecode cache contains an unexpected path: {entry}"
+				)
+		try:
+			for entry in entries:
+				entry.unlink()
+			cache.rmdir()
+		except OSError as exc:
+			raise LauncherUpdateError(
+				f"cannot remove launcher bytecode cache {cache}: {exc}"
+			) from exc
+
+
 def _self_check_release(release: Path, expected_version: str) -> None:
 	check = (
 		"import sys; "
--- a/ptlinux/system_identity.py
+++ b/ptlinux/system_identity.py
@@ -0,0 +1,83 @@
+"""Identify the host OS without losing the launcher's execution runtime."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+import re
+
+
+EXECUTION_OS_RELEASE = Path("/etc/os-release")
+CONTAINER_HOST_OS_RELEASE = Path("/run/host/os-release")
+STEAM_RUNTIME_OS_ID = "steamrt"
+
+
+@dataclass(frozen=True)
+class SystemIdentity:
+	"""OS releases visible to the launcher process and its container host."""
+
+	execution_release: dict[str, str]
+	host_release: dict[str, str] | None
+
+	@property
+	def operating_system_release(self) -> dict[str, str]:
+		return self.host_release or self.execution_release
+
+	@property
+	def running_inside_steam_runtime(self) -> bool:
+		return (
+			self.execution_release.get("ID", "").strip().casefold()
+			== STEAM_RUNTIME_OS_ID
+		)
+
+
+def detect_system_identity(
+	*,
+	execution_os_release: Path = EXECUTION_OS_RELEASE,
+	host_os_release: Path = CONTAINER_HOST_OS_RELEASE,
+) -> SystemIdentity:
+	"""Read the runtime OS and the standard container-host identity, if present."""
+
+	execution = read_os_release(execution_os_release)
+	host: dict[str, str] | None = None
+	if execution.get("ID", "").strip().casefold() == STEAM_RUNTIME_OS_ID:
+		candidate = read_os_release(host_os_release)
+		candidate_id = candidate.get("ID", "").strip().casefold()
+		if candidate_id and candidate_id != STEAM_RUNTIME_OS_ID:
+			host = candidate
+	return SystemIdentity(execution_release=execution, host_release=host)
+
+
+def detect_distribution_id() -> str:
+	"""Return the host distribution ID when a container exposes it."""
+
+	release = detect_system_identity().operating_system_release
+	return release.get("ID", "").strip().casefold()
+
+
+def is_running_inside_steam_runtime() -> bool:
+	"""Return whether the launcher process itself is inside Steam Runtime."""
+
+	return detect_system_identity().running_inside_steam_runtime
+
+
+def read_os_release(path: Path) -> dict[str, str]:
+	"""Read bounded freedesktop os-release fields without invoking host tools."""
+
+	try:
+		lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
+	except OSError:
+		return {}
+	result: dict[str, str] = {}
+	for line in lines:
+		if not line or line.startswith("#") or "=" not in line:
+			continue
+		key, raw = line.split("=", 1)
+		if not re.fullmatch(r"[A-Z][A-Z0-9_]*", key):
+			continue
+		value = raw.strip()
+		if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
+			value = value[1:-1]
+		value = value.replace(r"\n", " ").replace(r"\"", '"').replace(r"\\", "\\")
+		result[key] = value[:255]
+	return result
--- a/ptlinux/update_resolver.py
+++ b/ptlinux/update_resolver.py
@@ -0,0 +1,375 @@
+"""Semantic release versions and channel-aware update selection."""
+
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+import json
+import re
+from typing import Iterable, Sequence
+
+
+STABLE_CHANNEL = "stable"
+PRERELEASE_CHANNEL = "prerelease"
+SUPPORTED_CHANNELS = frozenset({STABLE_CHANNEL, PRERELEASE_CHANNEL})
+ACTIVE_STATUS = "active"
+WITHDRAWN_STATUS = "withdrawn"
+BLOCKED_STATUS = "blocked"
+SUPPORTED_RELEASE_STATUSES = frozenset(
+	{ACTIVE_STATUS, WITHDRAWN_STATUS, BLOCKED_STATUS}
+)
+
+_VERSION_PATTERN = re.compile(
+	r"(?P<major>0|[1-9][0-9]*)\."
+	r"(?P<minor>0|[1-9][0-9]*)\."
+	r"(?P<patch>0|[1-9][0-9]*)"
+	r"(?:-(?P<label>beta|rc)\.(?P<number>0|[1-9][0-9]*))?\Z"
+)
+
+
+class ReleaseModelError(ValueError):
+	"""Raised when release metadata violates the supported release model."""
+
+
+@dataclass(frozen=True)
+class SemanticVersion:
+	"""The stable, beta, and release-candidate version subset used by releases."""
+
+	major: int
+	minor: int
+	patch: int
+	prerelease_label: str | None = None
+	prerelease_number: int | None = None
+
+	@classmethod
+	def parse(cls, value: str) -> SemanticVersion:
+		if not isinstance(value, str):
+			raise ReleaseModelError("version must be a string")
+		match = _VERSION_PATTERN.fullmatch(value)
+		if match is None:
+			raise ReleaseModelError(
+				"version must use MAJOR.MINOR.PATCH, "
+				"MAJOR.MINOR.PATCH-beta.N, or MAJOR.MINOR.PATCH-rc.N"
+			)
+		label = match.group("label")
+		number_text = match.group("number")
+		return cls(
+			major=int(match.group("major")),
+			minor=int(match.group("minor")),
+			patch=int(match.group("patch")),
+			prerelease_label=label,
+			prerelease_number=int(number_text) if number_text is not None else None,
+		)
+
+	@classmethod
+	def from_tag(cls, tag: str) -> SemanticVersion:
+		if not isinstance(tag, str) or not tag.startswith("v"):
+			raise ReleaseModelError("release tag must start with 'v'")
+		return cls.parse(tag[1:])
+
+	@property
+	def is_prerelease(self) -> bool:
+		return self.prerelease_label is not None
+
+	@property
+	def channel(self) -> str:
+		return PRERELEASE_CHANNEL if self.is_prerelease else STABLE_CHANNEL
+
+	@property
+	def tag(self) -> str:
+		return f"v{self}"
+
+	@property
+	def target(self) -> tuple[int, int, int]:
+		return self.major, self.minor, self.patch
+
+	def _ordering_key(self) -> tuple[int, int, int, int, int, int]:
+		if self.prerelease_label is None:
+			return self.major, self.minor, self.patch, 1, 0, 0
+		label_rank = 0 if self.prerelease_label == "beta" else 1
+		assert self.prerelease_number is not None
+		return (
+			self.major,
+			self.minor,
+			self.patch,
+			0,
+			label_rank,
+			self.prerelease_number,
+		)
+
+	def __lt__(self, other: object) -> bool:
+		if not isinstance(other, SemanticVersion):
+			return NotImplemented
+		return self._ordering_key() < other._ordering_key()
+
+	def __le__(self, other: object) -> bool:
+		if not isinstance(other, SemanticVersion):
+			return NotImplemented
+		return self._ordering_key() <= other._ordering_key()
+
+	def __gt__(self, other: object) -> bool:
+		if not isinstance(other, SemanticVersion):
+			return NotImplemented
+		return self._ordering_key() > other._ordering_key()
+
+	def __ge__(self, other: object) -> bool:
+		if not isinstance(other, SemanticVersion):
+			return NotImplemented
+		return self._ordering_key() >= other._ordering_key()
+
+	def __str__(self) -> str:
+		value = f"{self.major}.{self.minor}.{self.patch}"
+		if self.prerelease_label is not None:
+			value += f"-{self.prerelease_label}.{self.prerelease_number}"
+		return value
+
+
+@dataclass(frozen=True)
+class ReleaseRecord:
+	"""The resolver-facing fields shared by channel indexes and release history."""
+
+	version: str
+	channel: str
+	status: str = ACTIVE_STATUS
+	tag: str | None = None
+	manifest_url: str | None = None
+	manifest_signature_url: str | None = None
+	release_date: str | None = None
+	release_notes_url: str | None = None
+	compatibility: object | None = None
+
+	@property
+	def semantic_version(self) -> SemanticVersion:
+		return SemanticVersion.parse(self.version)
+
+	def validate(self) -> ReleaseRecord:
+		parsed = self.semantic_version
+		validate_channel(self.channel)
+		if parsed.channel != self.channel:
+			raise ReleaseModelError(
+				f"version {self.version} belongs to {parsed.channel}, not {self.channel}"
+			)
+		if self.status not in SUPPORTED_RELEASE_STATUSES:
+			raise ReleaseModelError(f"unsupported release status: {self.status}")
+		if self.tag is not None and self.tag != parsed.tag:
+			raise ReleaseModelError(
+				f"release tag {self.tag!r} does not match version {self.version}"
+			)
+		return self
+
+
+class UpdateResolver:
+	"""Centralize release eligibility for updates, switching, and rollback."""
+
+	def __init__(self, releases: Iterable[ReleaseRecord]) -> None:
+		validated: dict[SemanticVersion, ReleaseRecord] = {}
+		for release in releases:
+			release.validate()
+			parsed = release.semantic_version
+			if parsed in validated:
+				raise ReleaseModelError(f"duplicate release version: {release.version}")
+			validated[parsed] = release
+		self._releases = validated
+
+	def eligible_releases(
+		self,
+		selected_channel: str,
+		*,
+		include_inactive: bool = False,
+	) -> tuple[ReleaseRecord, ...]:
+		validate_channel(selected_channel)
+		eligible = []
+		for parsed, release in self._releases.items():
+			if not include_inactive and release.status != ACTIVE_STATUS:
+				continue
+			if selected_channel == STABLE_CHANNEL and parsed.is_prerelease:
+				continue
+			eligible.append(release)
+		eligible.sort(key=lambda item: item.semantic_version, reverse=True)
+		return tuple(eligible)
+
+	def latest_eligible(self, selected_channel: str) -> ReleaseRecord | None:
+		eligible = self.eligible_releases(selected_channel)
+		return eligible[0] if eligible else None
+
+	def update_candidate(
+		self,
+		installed_version: str,
+		selected_channel: str,
+	) -> ReleaseRecord | None:
+		installed = SemanticVersion.parse(installed_version)
+		latest = self.latest_eligible(selected_channel)
+		if latest is None or latest.semantic_version <= installed:
+			return None
+		return latest
+
+	def rollback_candidates(
+		self,
+		installed_version: str,
+		selected_channel: str,
+		*,
+		show_prereleases: bool | None = None,
+		include_inactive: bool = False,
+	) -> tuple[ReleaseRecord, ...]:
+		installed = SemanticVersion.parse(installed_version)
+		validate_channel(selected_channel)
+		include_prereleases = (
+			selected_channel == PRERELEASE_CHANNEL
+			if show_prereleases is None
+			else show_prereleases
+		)
+		candidates = []
+		for release in self._releases.values():
+			parsed = release.semantic_version
+			if not include_inactive and release.status != ACTIVE_STATUS:
+				continue
+			if parsed >= installed:
+				continue
+			if parsed.is_prerelease and not include_prereleases:
+				continue
+			candidates.append(release)
+		candidates.sort(key=lambda item: item.semantic_version, reverse=True)
+		return tuple(candidates)
+
+
+def validate_channel(channel: str) -> str:
+	if channel not in SUPPORTED_CHANNELS:
+		raise ReleaseModelError(
+			f"channel must be one of: {', '.join(sorted(SUPPORTED_CHANNELS))}"
+		)
+	return channel
+
+
+def validate_channel_version(channel: str, version: str) -> SemanticVersion:
+	validate_channel(channel)
+	parsed = SemanticVersion.parse(version)
+	if parsed.channel != channel:
+		raise ReleaseModelError(
+			f"{channel} channel cannot publish version {version}; "
+			f"the version belongs to {parsed.channel}"
+		)
+	return parsed
+
+
+def validate_release_progression(
+	channel: str,
+	version: str,
+	tags: Iterable[str],
+) -> SemanticVersion:
+	"""Reject a release that would move its channel index backwards."""
+
+	target = validate_channel_version(channel, version)
+	relevant: list[SemanticVersion] = []
+	for tag in tags:
+		try:
+			candidate = SemanticVersion.from_tag(tag)
+		except ReleaseModelError:
+			continue
+		if channel == STABLE_CHANNEL and candidate.is_prerelease:
+			continue
+		relevant.append(candidate)
+	if relevant:
+		latest = max(relevant)
+		if target <= latest:
+			raise ReleaseModelError(
+				f"{channel} release {version} must be newer than existing "
+				f"release {latest}"
+			)
+	return target
+
+
+def previous_release_tag(
+	channel: str,
+	version: str,
+	tags: Iterable[str],
+) -> str | None:
+	"""Select the changelog base required by the two release channels."""
+
+	target = validate_channel_version(channel, version)
+	parsed_tags: list[SemanticVersion] = []
+	for tag in tags:
+		try:
+			candidate = SemanticVersion.from_tag(tag)
+		except ReleaseModelError:
+			continue
+		if candidate < target:
+			parsed_tags.append(candidate)
+	if channel == STABLE_CHANNEL:
+		stable = [candidate for candidate in parsed_tags if not candidate.is_prerelease]
+		return max(stable).tag if stable else None
+	matching_prereleases = [
+		candidate
+		for candidate in parsed_tags
+		if candidate.is_prerelease and candidate.target == target.target
+	]
+	if matching_prereleases:
+		return max(matching_prereleases).tag
+	stable = [candidate for candidate in parsed_tags if not candidate.is_prerelease]
+	return max(stable).tag if stable else None
+
+
+def release_metadata_from_tag(tag: str) -> dict[str, object]:
+	version = SemanticVersion.from_tag(tag)
+	return {
+		"tag": tag,
+		"version": str(version),
+		"channel": version.channel,
+		"prerelease": version.is_prerelease,
+	}
+
+
+def _build_parser() -> argparse.ArgumentParser:
+	parser = argparse.ArgumentParser(description=__doc__)
+	subparsers = parser.add_subparsers(dest="command", required=True)
+	validate = subparsers.add_parser("validate", help="validate a channel/version pair")
+	validate.add_argument("channel", choices=sorted(SUPPORTED_CHANNELS))
+	validate.add_argument("version")
+	inspect_tag = subparsers.add_parser("inspect-tag", help="derive release metadata")
+	inspect_tag.add_argument("tag")
+	previous = subparsers.add_parser("previous-tag", help="select a changelog base")
+	previous.add_argument("channel", choices=sorted(SUPPORTED_CHANNELS))
+	previous.add_argument("version")
+	previous.add_argument("tags", nargs="*")
+	progression = subparsers.add_parser(
+		"validate-next",
+		help="validate that a release advances its channel",
+	)
+	progression.add_argument("channel", choices=sorted(SUPPORTED_CHANNELS))
+	progression.add_argument("version")
+	progression.add_argument("tags", nargs="*")
+	return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+	arguments = _build_parser().parse_args(argv)
+	try:
+		if arguments.command == "validate":
+			validate_channel_version(arguments.channel, arguments.version)
+			return 0
+		if arguments.command == "inspect-tag":
+			print(json.dumps(release_metadata_from_tag(arguments.tag), sort_keys=True))
+			return 0
+		if arguments.command == "previous-tag":
+			selected = previous_release_tag(
+				arguments.channel,
+				arguments.version,
+				arguments.tags,
+			)
+			if selected is not None:
+				print(selected)
+			return 0
+		if arguments.command == "validate-next":
+			validate_release_progression(
+				arguments.channel,
+				arguments.version,
+				arguments.tags,
+			)
+			return 0
+	except ReleaseModelError as exc:
+		parser_error = _build_parser()
+		parser_error.error(str(exc))
+	raise AssertionError(f"unhandled command: {arguments.command}")
+
+
+if __name__ == "__main__":
+	raise SystemExit(main())
--- a/ptlinux/window_diagnostics.py
+++ b/ptlinux/window_diagnostics.py
@@ -0,0 +1,223 @@
+"""Bounded structured diagnostics for the two game-window supervisors."""
+
+from __future__ import annotations
+
+from collections import Counter
+from collections.abc import Callable, Mapping, Sequence
+import json
+import math
+import os
+import re
+import sys
+import time
+
+
+WINDOW_DIAGNOSTICS_ENV = "PTLINUX_WINDOW_DIAGNOSTICS"
+WINDOW_DIAGNOSTICS_SESSION_ENV = "PTLINUX_WINDOW_DIAGNOSTICS_SESSION"
+WINDOW_DIAGNOSTIC_PREFIX = "PTLINUX_WINDOW_DIAGNOSTIC "
+WINDOW_DIAGNOSTIC_EVENT_LIMIT = 12
+WINDOW_DIAGNOSTIC_LINE_LIMIT = 4096
+WINDOW_DIAGNOSTIC_COMPONENTS = frozenset({"game_window", "gamescope_host"})
+_DIAGNOSTIC_NAME_PATTERN = re.compile(r"[a-z][a-z0-9_]{0,63}\Z")
+_SESSION_ID_PATTERN = re.compile(r"[0-9a-f]{32}\Z")
+_TRUE_VALUES = frozenset({"1", "true", "yes"})
+_MAX_DETAILS_DEPTH = 4
+_MAX_DETAILS_ITEMS = 48
+_MAX_STRING_CHARACTERS = 256
+
+
+def window_diagnostics_enabled(
+	environment: Mapping[str, str] | None = None,
+) -> bool:
+	env = os.environ if environment is None else environment
+	return env.get(WINDOW_DIAGNOSTICS_ENV, "").strip().casefold() in _TRUE_VALUES
+
+
+class WindowDiagnosticRecorder:
+	"""Emit a few samples while retaining aggregate counts for session shutdown."""
+
+	def __init__(
+		self,
+		component: str,
+		*,
+		environment: Mapping[str, str] | None = None,
+		monotonic: Callable[[], float] | None = None,
+	) -> None:
+		if component not in WINDOW_DIAGNOSTIC_COMPONENTS:
+			raise ValueError(f"unsupported window diagnostic component: {component}")
+		env = os.environ if environment is None else environment
+		session_id = env.get(WINDOW_DIAGNOSTICS_SESSION_ENV, "").strip().casefold()
+		self.component = component
+		self.enabled = window_diagnostics_enabled(env)
+		self.session_id = (
+			session_id if _SESSION_ID_PATTERN.fullmatch(session_id) is not None else None
+		)
+		self._monotonic = monotonic or time.monotonic
+		self._started_at = self._monotonic()
+		self._counts: Counter[str] = Counter()
+		self._transitions: dict[str, object] = {}
+		self._sequence = 0
+		self._emitted = 0
+		self._closed = False
+
+	@property
+	def accepts_samples(self) -> bool:
+		return (
+			self.enabled
+			and not self._closed
+			and self._emitted < WINDOW_DIAGNOSTIC_EVENT_LIMIT
+		)
+
+	def record(self, diagnostic: str, **details: object) -> None:
+		if self._closed or _DIAGNOSTIC_NAME_PATTERN.fullmatch(diagnostic) is None:
+			return
+		self._sequence += 1
+		self._counts[diagnostic] += 1
+		if not self.enabled or self._emitted >= WINDOW_DIAGNOSTIC_EVENT_LIMIT:
+			return
+		self._emitted += 1
+		self._emit(diagnostic, self._sequence, details)
+
+	def transition(
+		self,
+		key: str,
+		value: object,
+		diagnostic: str,
+		**details: object,
+	) -> None:
+		"""Record one diagnostic only when the observed state changes."""
+
+		if self._closed or self._transitions.get(key, object()) == value:
+			return
+		self._transitions[key] = value
+		self.record(diagnostic, **details)
+
+	def close(self, **details: object) -> None:
+		if self._closed:
+			return
+		self._closed = True
+		if not self.enabled:
+			return
+		summary = {
+			**details,
+			"counts": dict(sorted(self._counts.items())),
+			"sampled_events": self._emitted,
+			"suppressed_events": max(0, self._sequence - self._emitted),
+		}
+		self._emit("session_summary", self._sequence + 1, summary)
+
+	def _emit(
+		self,
+		diagnostic: str,
+		sequence: int,
+		details: Mapping[str, object],
+	) -> None:
+		payload: dict[str, object] = {
+			"schema": 1,
+			"component": self.component,
+			"diagnostic": diagnostic,
+			"sequence": sequence,
+			"elapsed_ms": max(0, round((self._monotonic() - self._started_at) * 1000)),
+			"details": _bounded_value(details, depth=0),
+		}
+		if self.session_id is not None:
+			payload["session_id"] = self.session_id
+		try:
+			encoded = json.dumps(
+				payload,
+				sort_keys=True,
+				separators=(",", ":"),
+				ensure_ascii=True,
+			)
+			if len(encoded.encode("utf-8")) > WINDOW_DIAGNOSTIC_LINE_LIMIT:
+				payload["details"] = {"payload_truncated": True}
+				encoded = json.dumps(
+					payload,
+					sort_keys=True,
+					separators=(",", ":"),
+					ensure_ascii=True,
+				)
+			print(
+				f"{WINDOW_DIAGNOSTIC_PREFIX}{encoded}",
+				file=sys.stderr,
+				flush=True,
+			)
+		except (OSError, TypeError, ValueError):
+			return
+
+
+def parse_window_diagnostic(line: str) -> dict[str, object] | None:
+	"""Parse only launcher-generated, bounded window diagnostic records."""
+
+	marker_index = line.find(WINDOW_DIAGNOSTIC_PREFIX)
+	if marker_index < 0:
+		return None
+	encoded = line[marker_index + len(WINDOW_DIAGNOSTIC_PREFIX):].strip()
+	if not encoded or len(encoded.encode("utf-8")) > WINDOW_DIAGNOSTIC_LINE_LIMIT:
+		return None
+	try:
+		payload = json.loads(encoded)
+	except (json.JSONDecodeError, UnicodeError):
+		return None
+	if not isinstance(payload, dict) or payload.get("schema") != 1:
+		return None
+	component = payload.get("component")
+	diagnostic = payload.get("diagnostic")
+	sequence = payload.get("sequence")
+	elapsed_ms = payload.get("elapsed_ms")
+	details = payload.get("details")
+	session_id = payload.get("session_id")
+	if (
+		not isinstance(component, str)
+		or component not in WINDOW_DIAGNOSTIC_COMPONENTS
+		or not isinstance(diagnostic, str)
+		or _DIAGNOSTIC_NAME_PATTERN.fullmatch(diagnostic) is None
+		or not isinstance(sequence, int)
+		or isinstance(sequence, bool)
+		or sequence < 1
+		or not isinstance(elapsed_ms, int)
+		or isinstance(elapsed_ms, bool)
+		or elapsed_ms < 0
+		or not isinstance(details, dict)
+		or (
+			session_id is not None
+			and (
+				not isinstance(session_id, str)
+				or _SESSION_ID_PATTERN.fullmatch(session_id) is None
+			)
+		)
+	):
+		return None
+	return {
+		"schema": 1,
+		"component": component,
+		"diagnostic": diagnostic,
+		"sequence": sequence,
+		"elapsed_ms": elapsed_ms,
+		"details": _bounded_value(details, depth=0),
+		**({"session_id": session_id} if isinstance(session_id, str) else {}),
+	}
+
+
+def _bounded_value(value: object, *, depth: int) -> object:
+	if depth >= _MAX_DETAILS_DEPTH:
+		return "<max-depth>"
+	if value is None or isinstance(value, (bool, int)):
+		return value
+	if isinstance(value, float):
+		return round(value, 3) if math.isfinite(value) else None
+	if isinstance(value, str):
+		return value[:_MAX_STRING_CHARACTERS]
+	if isinstance(value, Mapping):
+		result: dict[str, object] = {}
+		for index, (raw_key, item) in enumerate(value.items()):
+			if index >= _MAX_DETAILS_ITEMS:
+				result["_truncated"] = True
+				break
+			key = str(raw_key)[:64]
+			result[key] = _bounded_value(item, depth=depth + 1)
+		return result
+	if isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray)):
+		items = list(value[:_MAX_DETAILS_ITEMS])
+		return [_bounded_value(item, depth=depth + 1) for item in items]
+	return f"<{type(value).__name__}>"