pristontale-eu-linux/ptlinux/overlay.py

"""Updater-safe runtime view with one verified normalized scene cache."""

from __future__ import annotations

from dataclasses import asdict, dataclass
from datetime import datetime
import hashlib
import json
import os
from pathlib import Path
import shutil
from typing import Mapping
import uuid

from ptdiag.assbin import (
	SceneInspection,
	inspect_wrapped_assbin,
	normalize_wrapped_assbin_textures,
)

from .paths import AppPaths


WHISPERING_VALE_RELATIVE = Path("game/maps/WhisperingVale.rovug")
KNOWN_SOURCE_SHA256 = (
	"1c9655945882b6b2189bbc6961018fa84295f895330987bad34e8428e9b5ce90"
)
TEXTURE_INDICES = (37, 38)
KNOWN_NORMALIZED_SCANLINES_SHA256 = {
	37: "8fcc54e8d30fbfb725bb4c917a123e4f28a228397b4939d2fcfc7e8f3edc84f5",
	38: "a8301700ee76b0cbf9d974c1395093623390244f2f92658b352a1c3e19e20f60",
}
COMPATIBILITY_BINARY_HASHES = {
	"Game.exe": "3cc0980aac0f65b379ff340ec61750498fb9f620fb6c88aa66281e8ab27023a5",
	"game.dll": "7ea85b19f452962a39c91f967cad22495f066b2d52e8ac37f4e234463d5c4f09",
}
CURRENT_V21_BINARY_HASHES = {
	"Game.exe": "2bc3edd26c5f6d7ae9bbdc23967c5aefd374bdf934fcdfa2ed42c6442e550837",
	"game.dll": "4b4e5c28e2bfdaa8d36c8341fad27becfd03faf63aa8ea874f5d43c5b1961868",
}
PATCHED_V21_GAME_DLL_SHA256 = (
	"0be11b06ccf9be747e32b195c8d5980062f446d7d3a0a662ebfe0e602a01c538"
)
RETIRED_POST_AUTH_GAME_DLL_SHA256 = (
	"93cd27284eb294753da1111c99bfc9689f82b6d4a59302aee9fdbd8cb52452dd"
)
RUNTIME_VIEW_CONFIGURATION_VERSION = 3
WRITABLE_ASSET_SUFFIXES = frozenset({".grx_img"})
WINDOWED_GAME_CONFIG_VALUES = (
	(b"windowed", b"Windowed", b"1"),
	(b"startmaximized", b"StartMaximized", b"False"),
)


class OverlayError(RuntimeError):
	"""Raised when a safe runtime view cannot be built."""


@dataclass(frozen=True)
class RuntimeFileIdentity:
	"""Cheap change detector for one previously hash-verified runtime input."""

	name: str
	device: int
	inode: int
	mode: int
	size: int
	mtime_ns: int
	ctime_ns: int


@dataclass(frozen=True)
class WritableAssetInventory:
	"""Source identities for asset formats the client is known to modify."""

	files: tuple[RuntimeFileIdentity, ...]
	bytes_total: int
	fingerprint: str


@dataclass(frozen=True)
class OverlayResult:
	source_game: str
	runtime_view: str
	source_scene_sha256: str
	runtime_scene_sha256: str
	scene_action: str
	binary_action: str
	game_exe_sha256: str
	game_dll_sha256: str
	input_identities: tuple[RuntimeFileIdentity, ...]
	runtime_scene_identity: RuntimeFileIdentity
	created_at: str
	links_created: int
	writable_asset_count: int = 0
	writable_asset_bytes: int = 0
	writable_asset_source_fingerprint: str = ""


@dataclass(frozen=True)
class CompatibilityCacheResult:
	source: str
	destination: str
	game_exe_sha256: str
	game_dll_sha256: str
	created_at: str


@dataclass(frozen=True)
class RuntimeCompatibility:
	"""Read-only compatibility result for the managed game currently on disk."""

	compatible: bool
	source_game: str
	source_scene_sha256: str | None
	scene_action: str | None
	binary_action: str | None
	reason: str | None


def inspect_runtime_compatibility(
	paths: AppPaths,
	source_game: Path,
) -> RuntimeCompatibility:
	"""Check whether ``prepare_runtime_view`` can recognize the installed release.

	This preflight is intentionally read-only: it never normalizes a scene,
	builds the verified runtime DLL, prepares save data, or replaces the runtime
	view. Preparation reuses a prior authoritative result only while every input
	identity and runtime-view link remains unchanged; otherwise it repeats the
	full checks before launch.
	"""

	requested_source = source_game.expanduser()
	source_scene_sha256 = None
	scene_action = None
	binary_action = None
	try:
		source = requested_source.resolve(strict=True)
		_validate_game_tree(source)
		cached = _cached_runtime_compatibility(paths, source)
		if cached is not None:
			return cached
		source_scene = source / WHISPERING_VALE_RELATIVE
		inspection = inspect_wrapped_assbin(source_scene)
		source_scene_sha256 = inspection.file_sha256
		scene_action = _scene_action(inspection)
		binary_action = _inspect_runtime_binary_action(paths, source)
	except (OSError, ValueError, OverlayError) as exc:
		reason = str(exc).strip() or type(exc).__name__
		return RuntimeCompatibility(
			compatible=False,
			source_game=str(requested_source.resolve(strict=False)),
			source_scene_sha256=source_scene_sha256,
			scene_action=scene_action,
			binary_action=binary_action,
			reason=reason,
		)
	return RuntimeCompatibility(
		compatible=True,
		source_game=str(source),
		source_scene_sha256=source_scene_sha256,
		scene_action=scene_action,
		binary_action=binary_action,
		reason=None,
	)


def prepare_runtime_view(
	paths: AppPaths,
	source_game: Path,
) -> OverlayResult:
	"""Build a linked runtime view with private copies for writable asset caches."""

	source = source_game.expanduser().resolve(strict=True)
	_validate_game_tree(source)
	_validate_runtime_target(paths)
	game_save = _prepare_persistent_save(paths, source)
	game_config = _prepare_persistent_game_config(paths, source)
	writable_assets = _writable_asset_inventory(source / "game")
	reused = _reuse_runtime_view(
		paths,
		source,
		game_save,
		game_config,
		writable_assets,
	)
	if reused is not None:
		return reused
	input_identities = _runtime_input_identities(paths, source)
	source_scene = source / WHISPERING_VALE_RELATIVE
	inspection = inspect_wrapped_assbin(source_scene)
	action = _scene_action(inspection)
	(
		binary_action,
		game_exe_source,
		game_dll_source,
		_source_binary_hashes,
	) = _runtime_game_binary_profile(
		paths,
		source,
	)
	build = paths.data / f".{paths.runtime_view.name}.build-{uuid.uuid4().hex}"
	backup = paths.data / f".{paths.runtime_view.name}.old-{uuid.uuid4().hex}"
	links_created = 0
	try:
		build.mkdir(parents=True, exist_ok=False)
		top_level_exceptions = {"game", "game.ini", "save"}
		for name, binary_source in (
			("Game.exe", game_exe_source),
			("game.dll", game_dll_source),
		):
			if binary_source != source / name:
				top_level_exceptions.add(name)
		links_created += _link_level(
			source,
			build,
			except_names=top_level_exceptions,
		)
		(build / "save").symlink_to(game_save, target_is_directory=True)
		links_created += 1
		(build / "game.ini").symlink_to(game_config)
		links_created += 1
		if game_exe_source != source / "Game.exe":
			(build / "Game.exe").symlink_to(game_exe_source)
			links_created += 1
		if game_dll_source != source / "game.dll":
			(build / "game.dll").symlink_to(game_dll_source)
			links_created += 1
		game_source = source / "game"
		game_target = build / "game"
		game_target.mkdir()
		links_created += _build_runtime_game_tree(
			game_source,
			game_target,
			writable_assets,
		)
		runtime_scene = build / WHISPERING_VALE_RELATIVE
		if action == "normalize-known-profile":
			normalization = normalize_wrapped_assbin_textures(
				source_scene,
				runtime_scene,
				TEXTURE_INDICES,
			)
			validate_scene_normalization(normalization)
		else:
			shutil.copy2(source_scene, runtime_scene)
		if _runtime_input_identities(paths, source) != input_identities:
			raise OverlayError(
				"game files changed while the managed runtime view was being prepared"
			)
		if _writable_asset_inventory(game_source) != writable_assets:
			raise OverlayError(
				"writable game assets changed while the managed runtime view was being prepared"
			)

		if paths.runtime_view.exists() or paths.runtime_view.is_symlink():
			paths.runtime_view.replace(backup)
		build.replace(paths.runtime_view)
		if backup.exists() or backup.is_symlink():
			_remove_runtime_tree(backup)
	except BaseException:
		_remove_runtime_tree(build)
		if backup.exists() and not paths.runtime_view.exists():
			backup.replace(paths.runtime_view)
		raise

	result = OverlayResult(
		source_game=str(source),
		runtime_view=str(paths.runtime_view),
		source_scene_sha256=inspection.file_sha256,
		runtime_scene_sha256=_sha256(paths.runtime_view / WHISPERING_VALE_RELATIVE),
		scene_action=action,
		binary_action=binary_action,
		game_exe_sha256=_sha256(paths.runtime_view / "Game.exe"),
		game_dll_sha256=_sha256(paths.runtime_view / "game.dll"),
		input_identities=input_identities,
		runtime_scene_identity=_runtime_file_identity(
			"runtime-scene",
			paths.runtime_view / WHISPERING_VALE_RELATIVE,
		),
		writable_asset_count=len(writable_assets.files),
		writable_asset_bytes=writable_assets.bytes_total,
		writable_asset_source_fingerprint=writable_assets.fingerprint,
		created_at=datetime.now().astimezone().isoformat(),
		links_created=links_created,
	)
	_write_json(
		paths.runtime_manifest,
		{
			"configuration_version": RUNTIME_VIEW_CONFIGURATION_VERSION,
			**asdict(result),
		},
	)
	return result


def cache_compatibility_binaries(
	paths: AppPaths,
	source_game: Path,
) -> CompatibilityCacheResult:
	source = source_game.expanduser().resolve(strict=True)
	hashes = _binary_hashes(source)
	if hashes != COMPATIBILITY_BINARY_HASHES:
		raise OverlayError(
			"compatibility source does not match the verified Game.exe/game.dll profile: "
			f"{source}"
		)
	target = paths.compatibility_binaries
	if target.exists() or target.is_symlink():
		if target.is_dir() and _binary_hashes(target) == COMPATIBILITY_BINARY_HASHES:
			return _compatibility_result(paths, source)
		raise OverlayError(f"compatibility cache has an unknown profile: {target}")
	paths.data.mkdir(parents=True, exist_ok=True)
	build = paths.data / f".{target.name}.build-{uuid.uuid4().hex}"
	try:
		build.mkdir(parents=False, exist_ok=False)
		for name in COMPATIBILITY_BINARY_HASHES:
			shutil.copy2(source / name, build / name)
		if _binary_hashes(build) != COMPATIBILITY_BINARY_HASHES:
			raise OverlayError("copied compatibility binaries failed hash verification")
		build.replace(target)
	except BaseException:
		_remove_runtime_tree(build)
		raise
	result = _compatibility_result(paths, source)
	_write_json(paths.compatibility_manifest, asdict(result))
	return result


def select_binary_action(
	cache_hashes: dict[str, str] | None,
	source_hashes: dict[str, str],
) -> str:
	if cache_hashes is not None:
		if cache_hashes != COMPATIBILITY_BINARY_HASHES:
			raise OverlayError("compatibility cache has an unknown binary profile")
	if source_hashes == COMPATIBILITY_BINARY_HASHES:
		return "verified-source-pair"
	if source_hashes == CURRENT_V21_BINARY_HASHES:
		return "verified-v21-publisher-pair"
	game_exe_sha256 = source_hashes.get("Game.exe")
	game_dll_sha256 = source_hashes.get("game.dll")
	if game_dll_sha256 in {
		PATCHED_V21_GAME_DLL_SHA256,
		RETIRED_POST_AUTH_GAME_DLL_SHA256,
	}:
		raise OverlayError(
			"a retired experimental game.dll is installed; run update --repair "
			"before launching"
		)
	known_game_exe_hashes = {
		CURRENT_V21_BINARY_HASHES["Game.exe"],
		COMPATIBILITY_BINARY_HASHES["Game.exe"],
	}
	known_game_dll_hashes = {
		CURRENT_V21_BINARY_HASHES["game.dll"],
		COMPATIBILITY_BINARY_HASHES["game.dll"],
	}
	if (
		game_exe_sha256 in known_game_exe_hashes
		or game_dll_sha256 in known_game_dll_hashes
	):
		raise OverlayError(
			"Game.exe and game.dll are from different verified releases; run "
			"update --repair before launching"
		)
	return "upstream-pair"


def _scene_action(inspection: SceneInspection) -> str:
	selected = []
	for index in TEXTURE_INDICES:
		if index >= len(inspection.embedded_textures):
			raise OverlayError(f"scene no longer has embedded texture {index}")
		selected.append(inspection.embedded_textures[index])
	depths = [texture.png.bit_depth if texture.png is not None else None for texture in selected]
	colors = [texture.png.color_type if texture.png is not None else None for texture in selected]
	return select_scene_action(inspection.file_sha256, depths, colors)


def select_scene_action(
	file_sha256: str,
	depths: list[int | None],
	colors: list[int | None],
) -> str:
	"""Choose a repair only from the exact source hash and decoded PNG profile."""

	if file_sha256 == KNOWN_SOURCE_SHA256 and depths == [16, 16] and colors == [2, 2]:
		return "normalize-known-profile"
	if depths == [8, 8] and colors == [2, 2]:
		return "upstream-compatible-copy"
	raise OverlayError(
		"WhisperingVale.rovug is not a supported release profile; refusing to "
		f"modify unknown textures (sha256={file_sha256}, "
		f"depths={depths}, colors={colors})"
	)


def validate_scene_normalization(report: Mapping[str, object]) -> None:
	"""Verify the known repair by decoded pixels, not zlib's encoded bytes.

	zlib is allowed to emit different valid DEFLATE streams across versions and
	compatible implementations.  The old whole-file hash therefore rejected a
	semantically identical scene on some distributions.  The source file remains
	pinned exactly, while these hashes pin the decoded scanlines produced for the
	two repaired textures.
	"""

	if report.get("source_sha256") != KNOWN_SOURCE_SHA256:
		raise OverlayError("normalized scene source did not match the verified release profile")
	verified = report.get("verified")
	if not isinstance(verified, bool) or not verified:
		raise OverlayError("normalized scene did not pass structural verification")
	if report.get("texture_indices") != list(TEXTURE_INDICES):
		raise OverlayError("normalized scene changed an unexpected texture set")

	changes = report.get("changes")
	if not isinstance(changes, list):
		raise OverlayError("normalized scene report did not contain texture changes")
	change_hashes: dict[int, str] = {}
	for change in changes:
		if not isinstance(change, dict):
			raise OverlayError("normalized scene report contained an invalid texture change")
		index = change.get("index")
		scanlines_sha256 = change.get("new_scanlines_sha256")
		if isinstance(index, bool) or not isinstance(index, int):
			raise OverlayError("normalized scene report contained an invalid texture index")
		if not isinstance(scanlines_sha256, str):
			raise OverlayError("normalized scene report omitted a decoded texture hash")
		if index in change_hashes:
			raise OverlayError("normalized scene report repeated a texture change")
		change_hashes[index] = scanlines_sha256

	if change_hashes != KNOWN_NORMALIZED_SCANLINES_SHA256:
		raise OverlayError(
			"normalized scene pixels did not match the verified release profile"
		)


def _validate_game_tree(source: Path) -> None:
	for relative in (
		Path("Game.exe"),
		Path("game.dll"),
		Path("game.ini"),
		WHISPERING_VALE_RELATIVE,
	):
		if not (source / relative).is_file():
			raise OverlayError(f"game installation is incomplete: missing {relative}")


def _validate_runtime_target(paths: AppPaths) -> None:
	data = paths.data.resolve()
	target = paths.runtime_view
	if target.parent.resolve() != data:
		raise OverlayError(f"runtime view is outside the application data root: {target}")
	if target == data:
		raise OverlayError("runtime view cannot be the application data root")


def _link_level(source: Path, target: Path, *, except_names: set[str]) -> int:
	count = 0
	for child in source.iterdir():
		if child.name in except_names:
			continue
		(target / child.name).symlink_to(child, target_is_directory=child.is_dir())
		count += 1
	return count


def _writable_asset_inventory(game_source: Path) -> WritableAssetInventory:
	"""Collect only formats observed writing cache headers during game sessions."""

	files: list[RuntimeFileIdentity] = []
	for directory, directory_names, file_names in os.walk(game_source, followlinks=False):
		parent = Path(directory)
		directory_names[:] = sorted(
			name for name in directory_names if not (parent / name).is_symlink()
		)
		for name in sorted(file_names):
			path = parent / name
			if path.suffix.casefold() not in WRITABLE_ASSET_SUFFIXES:
				continue
			if path.is_symlink() or not path.is_file():
				raise OverlayError(f"writable game asset is not a regular file: {path}")
			relative = path.relative_to(game_source).as_posix()
			files.append(
				_runtime_file_identity(relative, path, follow_symlinks=False)
			)
	files.sort(key=lambda item: item.name.casefold())
	digest = hashlib.sha256()
	for identity in files:
		digest.update(identity.name.encode("utf-8", errors="surrogateescape"))
		digest.update(b"\0")
		for value in (
			identity.device,
			identity.inode,
			identity.mode,
			identity.size,
			identity.mtime_ns,
			identity.ctime_ns,
		):
			digest.update(str(value).encode("ascii"))
			digest.update(b"\0")
	return WritableAssetInventory(
		files=tuple(files),
		bytes_total=sum(identity.size for identity in files),
		fingerprint=digest.hexdigest(),
	)


def _build_runtime_game_tree(
	game_source: Path,
	game_target: Path,
	inventory: WritableAssetInventory,
) -> int:
	"""Build a sparse writable view while linking every immutable subtree."""

	writable_files = {Path(identity.name) for identity in inventory.files}
	materialized_files = writable_files | {WHISPERING_VALE_RELATIVE.relative_to("game")}
	materialized_directories = _materialized_directories(materialized_files)

	def build_level(source: Path, target: Path, relative_parent: Path) -> int:
		links = 0
		for child in sorted(source.iterdir(), key=lambda path: path.name.casefold()):
			relative = relative_parent / child.name
			destination = target / child.name
			if relative == WHISPERING_VALE_RELATIVE.relative_to("game"):
				continue
			if relative in writable_files:
				if child.is_symlink() or not child.is_file():
					raise OverlayError(f"writable game asset is not a regular file: {child}")
				shutil.copy2(child, destination)
				destination.chmod(destination.stat().st_mode | 0o200)
				continue
			if (
				relative in materialized_directories
				and child.is_dir()
				and not child.is_symlink()
			):
				destination.mkdir()
				links += build_level(child, destination, relative)
				continue
			destination.symlink_to(child, target_is_directory=child.is_dir())
			links += 1
		return links

	return build_level(game_source, game_target, Path())


def _materialized_directories(files: set[Path]) -> set[Path]:
	directories: set[Path] = set()
	for path in files:
		parent = path.parent
		while parent != Path("."):
			directories.add(parent)
			parent = parent.parent
	return directories


def _prepare_persistent_save(paths: AppPaths, source: Path) -> Path:
	"""Migrate game-created settings and logs before replacing the runtime view."""

	target = paths.game_save
	if target.is_symlink():
		raise OverlayError(f"persistent game save must not be a symbolic link: {target}")
	if target.exists():
		if not target.is_dir():
			raise OverlayError(f"persistent game save is not a directory: {target}")
		_reject_save_symlinks(target)
		return target

	candidate = None
	for path in (paths.runtime_view / "save", source / "save"):
		if path.is_dir():
			candidate = path.resolve(strict=True)
			break
	if candidate is not None:
		selected = candidate
		allowed_roots = (paths.data.resolve(strict=False), source.resolve(strict=True))
		if not any(
			selected == root or selected.is_relative_to(root)
			for root in allowed_roots
		):
			raise OverlayError(f"refusing to migrate game save outside managed roots: {selected}")
		_reject_save_symlinks(selected)

	target.parent.mkdir(parents=True, exist_ok=True)
	staging = target.parent / f".{target.name}.build-{uuid.uuid4().hex}"
	try:
		if candidate is None:
			staging.mkdir(parents=False, exist_ok=False)
		else:
			shutil.copytree(candidate, staging, symlinks=True)
		staging.replace(target)
	except BaseException:
		_remove_runtime_tree(staging)
		raise
	return target


def _prepare_persistent_game_config(paths: AppPaths, source: Path) -> Path:
	"""Keep game settings writable while forcing a true windowed startup."""

	target = paths.game_config
	if target.is_symlink():
		raise OverlayError(f"persistent game config must not be a symbolic link: {target}")
	if target.exists() and not target.is_file():
		raise OverlayError(f"persistent game config is not a file: {target}")
	source_config = target if target.is_file() else source / "game.ini"
	try:
		content = source_config.read_bytes()
		mode = source_config.stat().st_mode & 0o777
	except OSError as exc:
		raise OverlayError(f"could not read game config {source_config}: {exc}") from exc
	windowed_content = _force_windowed_game_config(content)
	if source_config == target and windowed_content == content:
		return target

	target.parent.mkdir(parents=True, exist_ok=True)
	temporary = target.with_name(f".{target.name}.{uuid.uuid4().hex}.tmp")
	try:
		with temporary.open("xb") as handle:
			handle.write(windowed_content)
			handle.flush()
			os.fsync(handle.fileno())
		temporary.chmod(mode)
		temporary.replace(target)
	except BaseException:
		temporary.unlink(missing_ok=True)
		raise
	return target


def _force_windowed_game_config(content: bytes) -> bytes:
	"""Set only the two Screen keys that can trigger a desktop-sized backbuffer."""

	if b"\x00" in content:
		raise OverlayError("game.ini uses an unsupported encoding")
	newline = b"\r\n" if b"\r\n" in content else b"\n"
	desired = {
		key: (display_name, value)
		for key, display_name, value in WINDOWED_GAME_CONFIG_VALUES
	}
	result: list[bytes] = []
	in_screen = False
	found_screen = False
	found_settings: set[bytes] = set()

	def append_missing_settings() -> None:
		if result and not result[-1].endswith((b"\r", b"\n")):
			result[-1] += newline
		for key, display_name, value in WINDOWED_GAME_CONFIG_VALUES:
			if key in found_settings:
				continue
			result.append(display_name + b"=" + value + newline)
			found_settings.add(key)

	for line in content.splitlines(keepends=True):
		if line.endswith(b"\r\n"):
			body, line_ending = line[:-2], b"\r\n"
		elif line.endswith((b"\r", b"\n")):
			body, line_ending = line[:-1], line[-1:]
		else:
			body, line_ending = line, b""
		stripped = body.strip()
		if stripped.startswith(b"[") and b"]" in stripped:
			if in_screen:
				append_missing_settings()
			section_end = stripped.find(b"]")
			in_screen = stripped[1:section_end].strip().lower() == b"screen"
			found_screen = found_screen or in_screen
		if in_screen and b"=" in body and not stripped.startswith((b";", b"#")):
			equals = body.find(b"=")
			key = body[:equals].strip().lower()
			setting = desired.get(key)
			if setting is not None:
				leading_value_space = body[equals + 1 :]
				leading_value_space = leading_value_space[
					: len(leading_value_space) - len(leading_value_space.lstrip(b" \t"))
				]
				result.append(
					body[: equals + 1]
					+ leading_value_space
					+ setting[1]
					+ line_ending
				)
				found_settings.add(key)
				continue
		result.append(line)

	if in_screen:
		append_missing_settings()
	elif not found_screen:
		if result and not result[-1].endswith((b"\r", b"\n")):
			result[-1] += newline
		if result and result[-1].strip():
			result.append(newline)
		result.append(b"[Screen]" + newline)
		append_missing_settings()
	return b"".join(result)


def _reject_save_symlinks(root: Path) -> None:
	for directory, directory_names, file_names in os.walk(root, followlinks=False):
		parent = Path(directory)
		for name in (*directory_names, *file_names):
			path = parent / name
			if path.is_symlink():
				raise OverlayError(f"persistent game save contains a symbolic link: {path}")


def _runtime_file_identity(
	name: str,
	path: Path,
	*,
	follow_symlinks: bool = True,
) -> RuntimeFileIdentity:
	state = path.stat(follow_symlinks=follow_symlinks)
	return RuntimeFileIdentity(
		name=name,
		device=state.st_dev,
		inode=state.st_ino,
		mode=state.st_mode,
		size=state.st_size,
		mtime_ns=state.st_mtime_ns,
		ctime_ns=state.st_ctime_ns,
	)


def _runtime_input_identities(
	paths: AppPaths,
	source: Path,
) -> tuple[RuntimeFileIdentity, ...]:
	"""Snapshot inputs whose hashes established the cached runtime profile.

	Nanosecond mtime plus ctime/inode/device prevents ordinary edits, replacements,
	or timestamp restoration from bypassing the next authoritative hash check.
	"""

	identities = [
		_runtime_file_identity("source-game-exe", source / "Game.exe"),
		_runtime_file_identity("source-game-dll", source / "game.dll"),
		_runtime_file_identity("source-scene", source / WHISPERING_VALE_RELATIVE),
	]
	cache = paths.compatibility_binaries
	if cache.exists() or cache.is_symlink():
		identities.append(
			_runtime_file_identity(
				"compatibility-cache",
				cache,
				follow_symlinks=False,
			)
		)
		if cache.is_dir():
			for name in COMPATIBILITY_BINARY_HASHES:
				path = cache / name
				if path.exists() or path.is_symlink():
					identities.append(
						_runtime_file_identity(
							f"compatibility-cache-{name.casefold()}",
							path,
						)
					)
	return tuple(identities)


def _matching_cached_runtime_result(
	paths: AppPaths,
	source: Path,
	writable_assets: WritableAssetInventory | None = None,
) -> OverlayResult | None:
	result = _read_overlay_result(paths.runtime_manifest)
	if result is None:
		return None
	if result.source_game != str(source):
		return None
	if result.runtime_view != str(paths.runtime_view):
		return None
	if writable_assets is None:
		try:
			writable_assets = _writable_asset_inventory(source / "game")
		except OSError:
			return None
	if (
		result.writable_asset_count != len(writable_assets.files)
		or result.writable_asset_bytes != writable_assets.bytes_total
		or result.writable_asset_source_fingerprint != writable_assets.fingerprint
	):
		return None
	try:
		current_identities = _runtime_input_identities(paths, source)
	except OSError:
		return None
	if current_identities != result.input_identities:
		return None
	return result


def _cached_runtime_compatibility(
	paths: AppPaths,
	source: Path,
) -> RuntimeCompatibility | None:
	result = _matching_cached_runtime_result(paths, source)
	if result is None:
		return None
	return RuntimeCompatibility(
		compatible=True,
		source_game=str(source),
		source_scene_sha256=result.source_scene_sha256,
		scene_action=result.scene_action,
		binary_action=result.binary_action,
		reason=None,
	)


def _reuse_runtime_view(
	paths: AppPaths,
	source: Path,
	game_save: Path,
	game_config: Path,
	writable_assets: WritableAssetInventory,
) -> OverlayResult | None:
	result = _matching_cached_runtime_result(paths, source, writable_assets)
	if result is None:
		return None
	game_exe_source = source / "Game.exe"
	game_dll_source = source / "game.dll"
	_migrate_runtime_game_config_link(
		paths.runtime_view / "game.ini",
		source / "game.ini",
		game_config,
	)
	if not _runtime_layout_ready(
		paths,
		source,
		game_save,
		game_config,
		game_exe_source,
		game_dll_source,
		result.links_created,
		writable_assets,
	):
		return None
	runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE
	try:
		if (
			_runtime_file_identity("runtime-scene", runtime_scene)
			!= result.runtime_scene_identity
		):
			return None
		if _runtime_input_identities(paths, source) != result.input_identities:
			return None
	except OSError:
		return None
	return result


def _runtime_layout_ready(
	paths: AppPaths,
	source: Path,
	game_save: Path,
	game_config: Path,
	game_exe_source: Path,
	game_dll_source: Path,
	expected_links: int,
	writable_assets: WritableAssetInventory,
) -> bool:
	runtime = paths.runtime_view
	if runtime.is_symlink() or not runtime.is_dir():
		return False
	top_level_exceptions = {"game", "game.ini", "save"}
	top_level_special = {"game", "game.ini", "save"}
	for name, binary_source in (
		("Game.exe", game_exe_source),
		("game.dll", game_dll_source),
	):
		if binary_source != source / name:
			top_level_exceptions.add(name)
			top_level_special.add(name)
	top_level_ready, top_level_links = _linked_level_ready(
		source,
		runtime,
		except_names=top_level_exceptions,
		special_names=top_level_special,
	)
	if not top_level_ready:
		return False
	if not _symlink_points_to(runtime / "save", game_save):
		return False
	if not _symlink_points_to(runtime / "game.ini", game_config):
		return False
	if game_exe_source != source / "Game.exe" and not _symlink_points_to(
		runtime / "Game.exe",
		game_exe_source,
	):
		return False
	if game_dll_source != source / "game.dll" and not _symlink_points_to(
		runtime / "game.dll",
		game_dll_source,
	):
		return False

	game_ready, game_links = _runtime_game_tree_ready(
		source / "game",
		runtime / "game",
		writable_assets,
	)
	if not game_ready:
		return False
	actual_links = top_level_links + game_links + 2
	if game_exe_source != source / "Game.exe":
		actual_links += 1
	if game_dll_source != source / "game.dll":
		actual_links += 1
	return actual_links == expected_links


def _runtime_game_tree_ready(
	game_source: Path,
	game_target: Path,
	inventory: WritableAssetInventory,
) -> tuple[bool, int]:
	if game_target.is_symlink() or not game_target.is_dir():
		return False, 0
	writable_files = {
		Path(identity.name): identity.size
		for identity in inventory.files
	}
	protected_scene = WHISPERING_VALE_RELATIVE.relative_to("game")
	materialized_directories = _materialized_directories(
		set(writable_files) | {protected_scene}
	)

	def check_level(source: Path, target: Path, relative_parent: Path) -> tuple[bool, int]:
		try:
			source_children = {child.name: child for child in source.iterdir()}
			target_names = {child.name for child in target.iterdir()}
		except OSError:
			return False, 0
		if target_names != set(source_children):
			return False, 0
		links = 0
		for name, child in source_children.items():
			relative = relative_parent / name
			runtime_child = target / name
			if relative == protected_scene:
				if runtime_child.is_symlink() or not runtime_child.is_file():
					return False, 0
				continue
			writable_size = writable_files.get(relative)
			if writable_size is not None:
				try:
					if (
						runtime_child.is_symlink()
						or not runtime_child.is_file()
						or runtime_child.stat().st_size != writable_size
					):
						return False, 0
				except OSError:
					return False, 0
				continue
			if relative in materialized_directories:
				if runtime_child.is_symlink() or not runtime_child.is_dir():
					return False, 0
				ready, child_links = check_level(child, runtime_child, relative)
				if not ready:
					return False, 0
				links += child_links
				continue
			if not _symlink_points_to(runtime_child, child):
				return False, 0
			links += 1
		return True, links

	return check_level(game_source, game_target, Path())


def _linked_level_ready(
	source: Path,
	target: Path,
	*,
	except_names: set[str],
	special_names: set[str],
) -> tuple[bool, int]:
	try:
		source_children = {
			child.name: child
			for child in source.iterdir()
			if child.name not in except_names
		}
		target_names = {child.name for child in target.iterdir()}
	except OSError:
		return False, 0
	if target_names != set(source_children) | special_names:
		return False, 0
	for name, child in source_children.items():
		if not _symlink_points_to(target / name, child):
			return False, 0
	return True, len(source_children)


def _symlink_points_to(link: Path, expected_target: Path) -> bool:
	if not link.is_symlink():
		return False
	try:
		return link.resolve(strict=True) == expected_target.resolve(strict=True)
	except (OSError, RuntimeError):
		return False


def _migrate_runtime_game_config_link(
	link: Path,
	source_config: Path,
	persistent_config: Path,
) -> bool:
	"""Atomically redirect the previous source link without rebuilding the view."""

	if _symlink_points_to(link, persistent_config):
		return True
	if not _symlink_points_to(link, source_config):
		return False
	temporary = link.with_name(f".{link.name}.{uuid.uuid4().hex}.tmp")
	try:
		temporary.symlink_to(persistent_config)
		temporary.replace(link)
	except OSError:
		temporary.unlink(missing_ok=True)
		return False
	return _symlink_points_to(link, persistent_config)


def _read_overlay_result(path: Path) -> OverlayResult | None:
	try:
		value = json.loads(path.read_text(encoding="utf-8"))
	except (OSError, UnicodeError, json.JSONDecodeError):
		return None
	if not isinstance(value, dict):
		return None
	if value.get("configuration_version") != RUNTIME_VIEW_CONFIGURATION_VERSION:
		return None
	string_fields = (
		"source_game",
		"runtime_view",
		"source_scene_sha256",
		"runtime_scene_sha256",
		"scene_action",
		"binary_action",
		"game_exe_sha256",
		"game_dll_sha256",
		"writable_asset_source_fingerprint",
		"created_at",
	)
	if not all(isinstance(value.get(name), str) for name in string_fields):
		return None
	links_created = value.get("links_created")
	if isinstance(links_created, bool) or not isinstance(links_created, int):
		return None
	if links_created < 0:
		return None
	writable_asset_count = value.get("writable_asset_count")
	writable_asset_bytes = value.get("writable_asset_bytes")
	if (
		isinstance(writable_asset_count, bool)
		or not isinstance(writable_asset_count, int)
		or writable_asset_count < 0
		or isinstance(writable_asset_bytes, bool)
		or not isinstance(writable_asset_bytes, int)
		or writable_asset_bytes < 0
	):
		return None
	input_values = value.get("input_identities")
	if not isinstance(input_values, list) or not input_values:
		return None
	input_identities: list[RuntimeFileIdentity] = []
	for input_value in input_values:
		identity = _parse_runtime_file_identity(input_value)
		if identity is None:
			return None
		input_identities.append(identity)
	input_names = {identity.name for identity in input_identities}
	if len(input_names) != len(input_identities) or not {
		"source-game-exe",
		"source-game-dll",
		"source-scene",
	}.issubset(input_names):
		return None
	runtime_scene_identity = _parse_runtime_file_identity(
		value.get("runtime_scene_identity")
	)
	if (
		runtime_scene_identity is None
		or runtime_scene_identity.name != "runtime-scene"
	):
		return None
	return OverlayResult(
		source_game=value["source_game"],
		runtime_view=value["runtime_view"],
		source_scene_sha256=value["source_scene_sha256"],
		runtime_scene_sha256=value["runtime_scene_sha256"],
		scene_action=value["scene_action"],
		binary_action=value["binary_action"],
		game_exe_sha256=value["game_exe_sha256"],
		game_dll_sha256=value["game_dll_sha256"],
		input_identities=tuple(input_identities),
		runtime_scene_identity=runtime_scene_identity,
		writable_asset_count=writable_asset_count,
		writable_asset_bytes=writable_asset_bytes,
		writable_asset_source_fingerprint=value[
			"writable_asset_source_fingerprint"
		],
		created_at=value["created_at"],
		links_created=links_created,
	)


def _parse_runtime_file_identity(value: object) -> RuntimeFileIdentity | None:
	if not isinstance(value, dict):
		return None
	name = value.get("name")
	if not isinstance(name, str) or not name:
		return None
	fields: dict[str, int] = {}
	for field in ("device", "inode", "mode", "size", "mtime_ns", "ctime_ns"):
		field_value = value.get(field)
		if (
			isinstance(field_value, bool)
			or not isinstance(field_value, int)
			or field_value < 0
		):
			return None
		fields[field] = field_value
	return RuntimeFileIdentity(
		name=name,
		device=fields["device"],
		inode=fields["inode"],
		mode=fields["mode"],
		size=fields["size"],
		mtime_ns=fields["mtime_ns"],
		ctime_ns=fields["ctime_ns"],
	)


def _runtime_game_binary_profile(
	paths: AppPaths,
	source: Path,
) -> tuple[str, Path, Path, dict[str, str]]:
	cache = paths.compatibility_binaries
	cache_hashes = _binary_hashes(cache) if cache.is_dir() else None
	source_hashes = _binary_hashes(source)
	action = select_binary_action(cache_hashes, source_hashes)
	return (
		action,
		source / "Game.exe",
		source / "game.dll",
		source_hashes,
	)


def _inspect_runtime_binary_action(
	paths: AppPaths,
	source: Path,
) -> str:
	"""Mirror publisher binary selection without writing any runtime cache."""

	cache = paths.compatibility_binaries
	cache_hashes = _binary_hashes(cache) if cache.is_dir() else None
	source_hashes = _binary_hashes(source)
	return select_binary_action(cache_hashes, source_hashes)


def _binary_hashes(directory: Path) -> dict[str, str]:
	return {
		name: _sha256(directory / name)
		for name in COMPATIBILITY_BINARY_HASHES
		if (directory / name).is_file()
	}


def _compatibility_result(paths: AppPaths, source: Path) -> CompatibilityCacheResult:
	return CompatibilityCacheResult(
		source=str(source),
		destination=str(paths.compatibility_binaries),
		game_exe_sha256=COMPATIBILITY_BINARY_HASHES["Game.exe"],
		game_dll_sha256=COMPATIBILITY_BINARY_HASHES["game.dll"],
		created_at=datetime.now().astimezone().isoformat(),
	)


def _remove_runtime_tree(path: Path) -> None:
	if path.is_symlink() or path.is_file():
		path.unlink(missing_ok=True)
	elif path.is_dir():
		shutil.rmtree(path)


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


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")
	temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
	temporary.replace(path)