pristontale-eu-linux/ptdiag/experiment.py

"""Controlled Proton 7 runtime experiments and result generation."""

from __future__ import annotations

from dataclasses import dataclass, replace
from datetime import datetime
import json
import os
from pathlib import Path, PurePosixPath
import shutil
import signal
import subprocess
import sys
import time
from typing import Any, Mapping, Sequence

from .config import AppConfig, project_root
from .filesystem import sha256_file, verify_known_game_hashes
from .logs import (
	LogFileState,
	discover_current_native_logs,
	parse_current_run_log,
	read_current_run_log_text,
	snapshot_native_logs,
)
from .ini import IniError, inspect_ini_value
from .performance import PerformanceSampler
from .prefix import (
	PrefixSafetyError,
	cleanup_dedicated_wineserver,
	validate_dedicated_prefix,
)
from .process import collect_environment
from .registry import (
	DesktopRegistrySnapshot,
	desktop_registry_restored,
	read_registry_string_values,
	snapshot_desktop_registry,
)
from .result import ResultDirectory, create_result_directory, write_json
from .runner import CommandLogger, CommandResult
from .testcase import TestIniOverride, TestModification, Testcase
from .trace import process_trace_shards


_EXPLORER_KEY = r"HKCU\Software\Wine\Explorer"
_DESKTOPS_KEY = r"HKCU\Software\Wine\Explorer\Desktops"
_WINDOWS_GAME_DIRECTORY = r"C:\Program Files (x86)\Pristontale EU"
_DISPOSABLE_DIRECTORY_NAME = "ptdiag-disposable"
_PERSISTENT_STATE_CACHE_SENTINEL = "@ptdiag-persistent"
_PERSISTENT_STATE_CACHE_DIRECTORY = "ptdiag-cache/dxvk-state"


@dataclass(frozen=True)
class ExperimentResult:
	result_directory: Path
	summary: dict[str, Any]
	exit_code: int


class ExperimentError(RuntimeError):
	"""Raised for a controlled experiment setup or lifecycle failure."""


def run_experiment(
	config: AppConfig,
	testcase: Testcase,
	*,
	dry_run: bool,
	results_root: Path | None = None,
) -> ExperimentResult:
	if testcase.modification is not None:
		return _run_modified_experiment(
			config,
			testcase,
			dry_run=dry_run,
			results_root=results_root,
		)
	return _run_standard_experiment(
		config,
		testcase,
		dry_run=dry_run,
		results_root=results_root,
	)


def _run_standard_experiment(
	config: AppConfig,
	testcase: Testcase,
	*,
	dry_run: bool,
	results_root: Path | None = None,
	prepared_result: ResultDirectory | None = None,
	started_at: datetime | None = None,
) -> ExperimentResult:
	started_at = started_at or datetime.now().astimezone()
	result = prepared_result or create_result_directory(
		results_root or project_root() / "results", testcase.name, started_at
	)
	logger = CommandLogger(result.commands_path)
	(result.path / "runner.log").touch(exist_ok=True)
	(result.path / "native.log").touch(exist_ok=True)
	environment = collect_environment(config)
	hashes = _collect_hashes(config, testcase)
	write_json(result.path / "hashes.json", {"files": hashes})

	desktop_name = _desktop_name(result.path)
	metadata: dict[str, Any] = {
		"test": testcase.name,
		"description": testcase.description,
		"started_at": started_at.isoformat(),
		"result_directory": str(result.path),
		"dry_run": dry_run,
		"testcase": testcase.to_dict(),
		"config_source": str(config.source),
		"desktop_name": desktop_name,
		"native_log_directory": str(config.paths.native_logs),
	}
	write_json(result.metadata_path, metadata)

	preflight_errors = _preflight(config, testcase, environment, hashes)
	baseline = snapshot_native_logs(config.paths.native_logs)
	metadata["old_log_baseline"] = _serialize_log_states(baseline)
	write_json(result.metadata_path, metadata)
	registry_before: DesktopRegistrySnapshot | None = None
	try:
		registry_before = snapshot_desktop_registry(config.proton7.prefix, desktop_name)
	except OSError as exc:
		preflight_errors.append(f"Could not snapshot dedicated-prefix registry: {exc}")

	launch_argv = _launch_argv(config, testcase, result.path)
	env_overrides = _runtime_environment(config, testcase, result.path)
	state_cache_summary = _state_cache_plan(config, env_overrides)
	dll_overrides = (
		_collect_dll_overrides(config.proton7.prefix)
		if testcase.capture.dll_overrides
		else {}
	)
	_write_text(
		result.path / "environment.txt",
		json.dumps(
			{
				"preflight": environment,
				"runtime_overrides": dict(sorted(env_overrides.items())),
				"dll_overrides": dll_overrides,
			},
			indent=2,
			sort_keys=True,
		),
	)
	metadata["dll_overrides"] = dll_overrides
	write_json(result.metadata_path, metadata)
	summary: dict[str, Any] = {
		"ok": False,
		"status": "preflight-failed" if preflight_errors else "prepared",
		"dry_run": dry_run,
		"test": testcase.name,
		"result_directory": str(result.path),
		"preflight": {
			"ok": not preflight_errors,
			"errors": preflight_errors,
		},
		"launch": {
			"argv": launch_argv,
			"cwd": str(config.paths.live_game),
			"env_overrides": dict(sorted(env_overrides.items())),
			"timeout_seconds": testcase.timeout_seconds,
		},
		"native_log": {
			"directory": str(config.paths.native_logs),
			"baseline": _serialize_log_states(baseline),
			"source": None,
			"current_run_bytes": 0,
		},
		"registry": {
			"temporary_change": testcase.display.virtual_desktop,
			"before": registry_before.to_dict() if registry_before else None,
			"after": None,
			"restored": None,
		},
		"dxvk_state_cache": state_cache_summary,
	}

	if preflight_errors:
		_finalize(result, summary)
		return ExperimentResult(result.path, summary, 2)
	assert registry_before is not None

	if dry_run:
		_dry_run_commands(
			config,
			testcase,
			logger,
			registry_before,
			launch_argv,
			env_overrides,
		)
		registry_after = snapshot_desktop_registry(config.proton7.prefix, desktop_name)
		summary["registry"]["after"] = registry_after.to_dict()
		summary["registry"]["restored"] = desktop_registry_restored(
			registry_before, registry_after
		)
		summary["status"] = "dry-run"
		summary["ok"] = True
		summary["review_gate"] = (
			"Phase C dry-run complete; review artifacts before a live Game.exe launch."
		)
		_finalize(result, summary)
		return ExperimentResult(result.path, summary, 0)

	runtime_exit = 1
	setup_complete = False
	try:
		if state_cache_summary is not None:
			state_cache_summary = _prepare_state_cache(
				config,
				env_overrides,
				logger,
			)
			summary["dxvk_state_cache"] = state_cache_summary
		_cleanup_checked(config, logger)
		if testcase.display.virtual_desktop:
			_run_checked_commands(
				logger,
				_registry_setup_commands(config, desktop_name, testcase.display.size),
				cwd=config.paths.live_game,
				env_overrides=_proton_environment(config),
			)
		setup_complete = True
		baseline = snapshot_native_logs(config.paths.native_logs)
		metadata["old_log_baseline"] = _serialize_log_states(baseline)
		write_json(result.metadata_path, metadata)
		runtime = _run_and_watch(
			config,
			testcase,
			result,
			logger,
			launch_argv,
			env_overrides,
			baseline,
		)
		summary["runtime"] = runtime
		summary["native_log"].update(runtime.pop("native_log"))
		runtime_exit = (
			0
			if runtime.get("failure_detected") or runtime.get("scene_import_survived")
			else 1
		)
		trigger_epoch = runtime.get("trigger_epoch")
		if testcase.capture.strace_files and isinstance(trigger_epoch, (int, float)):
			trace = process_trace_shards(
				result.path / "raw-strace",
				result.path,
				trigger_epoch=float(trigger_epoch),
				focus_before_seconds=testcase.capture.focus_before_seconds,
				focus_after_seconds=testcase.capture.focus_after_seconds,
				live_game=config.paths.live_game,
			)
			summary["trace"] = trace.to_dict()
		else:
			summary["trace"] = None
		if runtime.get("failure_detected"):
			summary["status"] = "captured-scene-failure"
		elif runtime.get("scene_import_survived"):
			summary["status"] = "scene-import-survived"
		else:
			summary["status"] = "no-trigger"
		summary["ok"] = bool(
			runtime.get("failure_detected") or runtime.get("scene_import_survived")
		)
	except (
		ExperimentError,
		OSError,
		ValueError,
		PrefixSafetyError,
		subprocess.SubprocessError,
	) as exc:
		summary["status"] = "runner-error"
		summary["error"] = str(exc)
		summary["ok"] = False
		runtime_exit = 2
	finally:
		cleanup_error: str | None = None
		try:
			_cleanup_checked(config, logger)
		except (ExperimentError, OSError, ValueError, PrefixSafetyError) as exc:
			cleanup_error = str(exc)
		if testcase.display.virtual_desktop and setup_complete:
			try:
				_run_checked_commands(
					logger,
					_registry_restore_commands(config, registry_before),
					cwd=config.paths.live_game,
					env_overrides=_proton_environment(config),
				)
			except ExperimentError as exc:
				cleanup_error = f"{cleanup_error}; {exc}" if cleanup_error else str(exc)
		try:
			registry_after = snapshot_desktop_registry(config.proton7.prefix, desktop_name)
			summary["registry"]["after"] = registry_after.to_dict()
			summary["registry"]["restored"] = desktop_registry_restored(
				registry_before, registry_after
			)
		except OSError as exc:
			cleanup_error = f"{cleanup_error}; {exc}" if cleanup_error else str(exc)
		if cleanup_error:
			summary["cleanup_error"] = cleanup_error
			summary["ok"] = False
			runtime_exit = 2
		if state_cache_summary is not None:
			state_cache_summary["after"] = _state_cache_file_snapshot(
				Path(str(state_cache_summary["cache_file"]))
			)
			summary["dxvk_state_cache"] = state_cache_summary

	_finalize(result, summary)
	return ExperimentResult(result.path, summary, runtime_exit)


def _run_modified_experiment(
	config: AppConfig,
	testcase: Testcase,
	*,
	dry_run: bool,
	results_root: Path | None,
) -> ExperimentResult:
	modification = testcase.modification
	assert modification is not None
	started_at = datetime.now().astimezone()
	result = create_result_directory(
		results_root or project_root() / "results", testcase.name, started_at
	)
	(result.path / "runner.log").touch(exist_ok=True)
	(result.path / "native.log").touch(exist_ok=True)
	logger = CommandLogger(result.commands_path)
	source_environment = collect_environment(config)
	source_hashes = _collect_hashes(config, testcase)
	source_errors = _preflight(config, testcase, source_environment, source_hashes)
	write_json(
		result.path / "source-preflight.json",
		{
			"environment": source_environment,
			"hashes": source_hashes,
			"ok": not source_errors,
			"errors": source_errors,
		},
	)
	disposable_game = (
		config.paths.app_root / _DISPOSABLE_DIRECTORY_NAME / result.path.name
	).absolute()
	modification_summary: dict[str, Any] = {
		"kind": modification.kind,
		"source": modification.source,
		"destination": modification.destination,
		"texture_indices": list(modification.texture_indices),
		"ini_overrides": [
			{
				"path": item.path,
				"section": item.section,
				"key": item.key,
				"expected": item.expected,
				"value": item.value,
			}
			for item in modification.ini_overrides
		],
		"disposable_game": str(disposable_game),
		"source_preflight_ok": not source_errors,
		"prepared": False,
		"prefix_link_restored": None,
		"disposable_removed": None,
	}
	write_json(result.path / "modification.json", modification_summary)
	if source_errors:
		summary = _setup_failure_summary(
			testcase, result, dry_run, source_errors, modification_summary
		)
		_finalize(result, summary)
		return ExperimentResult(result.path, summary, 2)

	try:
		source, live_destination = _resource_copy_paths(
			config.paths.live_game, modification
		)
		ini_preflight = _inspect_ini_overrides(
			config.paths.live_game,
			modification.ini_overrides,
		)
		for item in ini_preflight:
			if not item["matches_expected"]:
				raise ExperimentError(
					f"INI preflight mismatch for [{item['section']}] {item['key']}: "
					f"expected {item['expected']!r}, found {item['current']!r}"
				)
	except (OSError, ValueError, ExperimentError) as exc:
		summary = _setup_failure_summary(
			testcase, result, dry_run, [str(exc)], modification_summary
		)
		_finalize(result, summary)
		return ExperimentResult(result.path, summary, 2)

	modification_summary.update(
		{
			"source_path": str(source),
			"live_destination_path": str(live_destination),
			"live_destination_existed_before": live_destination.exists(),
			"source_sha256": sha256_file(source),
			"source_size": source.stat().st_size,
			"ini_preflight": ini_preflight,
		}
	)
	if dry_run:
		_log_modification_plan(
			logger,
			config,
			disposable_game,
			source,
			modification,
		)
		outcome = _run_standard_experiment(
			config,
			testcase,
			dry_run=True,
			prepared_result=result,
			started_at=started_at,
		)
		modification_summary["status"] = "planned-only"
		outcome.summary["modification"] = modification_summary
		write_json(result.path / "modification.json", modification_summary)
		_finalize(result, outcome.summary)
		return outcome

	prefix_link = (
		config.proton7.prefix / "drive_c/Program Files (x86)/Pristontale EU"
	)
	original_link_target: str | None = None
	link_changed = False
	disposable_created = False
	runtime_config = config
	outcome: ExperimentResult | None = None
	setup_error: str | None = None
	cleanup_errors: list[str] = []
	try:
		_cleanup_checked(config, logger)
		preparation = _prepare_disposable_game(
			config,
			result,
			logger,
			modification,
			disposable_game,
			source,
		)
		disposable_created = True
		modification_summary.update(preparation)
		original_link_target = _replace_game_link(
			prefix_link,
			disposable_game,
			config.paths.live_game,
			logger,
		)
		link_changed = True
		modification_summary.update(
			{
				"prefix_link": str(prefix_link),
				"prefix_link_original_target": original_link_target,
				"prefix_link_runtime_target": str(disposable_game),
			}
		)
		runtime_config = replace(
			config,
			paths=replace(
				config.paths,
				live_game=disposable_game,
				native_logs=disposable_game / "save/logs",
			),
		)
		modification_summary["prepared"] = True
		write_json(result.path / "modification.json", modification_summary)
		outcome = _run_standard_experiment(
			runtime_config,
			testcase,
			dry_run=False,
			prepared_result=result,
			started_at=started_at,
		)
	except (
		ExperimentError,
		OSError,
		ValueError,
		PrefixSafetyError,
		subprocess.SubprocessError,
	) as exc:
		setup_error = str(exc)
	finally:
		try:
			_cleanup_checked(runtime_config, logger)
		except (ExperimentError, OSError, ValueError, PrefixSafetyError) as exc:
			cleanup_errors.append(f"wineserver cleanup: {exc}")
		if disposable_created and disposable_game.exists() and modification.ini_overrides:
			try:
				modification_summary["ini_runtime_final"] = _inspect_ini_overrides(
					disposable_game,
					modification.ini_overrides,
				)
			except (ExperimentError, IniError, OSError) as exc:
				cleanup_errors.append(f"final INI capture: {exc}")
		if link_changed and original_link_target is not None:
			try:
				_restore_game_link(
					prefix_link,
					original_link_target,
					config.paths.live_game,
					logger,
				)
				modification_summary["prefix_link_restored"] = True
			except (ExperimentError, OSError) as exc:
				modification_summary["prefix_link_restored"] = False
				cleanup_errors.append(f"prefix link restore: {exc}")
		if disposable_created or disposable_game.exists():
			try:
				_remove_disposable_game(config, result, disposable_game, logger)
				modification_summary["disposable_removed"] = True
			except (ExperimentError, OSError) as exc:
				modification_summary["disposable_removed"] = False
				cleanup_errors.append(f"disposable cleanup: {exc}")

	if outcome is None:
		errors = [setup_error or "Modified experiment did not start", *cleanup_errors]
		summary = _setup_failure_summary(
			testcase, result, False, errors, modification_summary
		)
		exit_code = 2
	else:
		summary = outcome.summary
		exit_code = outcome.exit_code
		if setup_error:
			summary["status"] = "setup-error"
			summary["error"] = setup_error
			summary["ok"] = False
			exit_code = 2
		if cleanup_errors:
			summary["cleanup_error"] = "; ".join(cleanup_errors)
			summary["ok"] = False
			exit_code = 2
	summary["modification"] = modification_summary
	write_json(result.path / "modification.json", modification_summary)
	_finalize(result, summary)
	return ExperimentResult(result.path, summary, exit_code)


def _resource_copy_paths(
	live_game: Path,
	modification: TestModification,
) -> tuple[Path, Path]:
	if modification.kind not in {"copy-file", "normalize-assbin-textures"}:
		raise ExperimentError(f"Unsupported modification kind: {modification.kind}")
	root = live_game.resolve(strict=True)
	source_relative = _safe_resource_relative_path(modification.source)
	destination_relative = _safe_resource_relative_path(modification.destination)
	source = root.joinpath(*source_relative.parts).resolve(strict=True)
	try:
		source.relative_to(root)
	except ValueError as exc:
		raise ExperimentError("Modification source escapes the live game tree") from exc
	if not source.is_file():
		raise ExperimentError(f"Modification source is not a file: {source}")
	destination = root.joinpath(*destination_relative.parts)
	destination_parent = destination.parent.resolve(strict=True)
	try:
		destination_parent.relative_to(root)
	except ValueError as exc:
		raise ExperimentError("Modification destination escapes the live game tree") from exc
	if modification.kind == "copy-file" and (destination.exists() or destination.is_symlink()):
		raise ExperimentError(
			f"Live modification destination already exists; refusing experiment: {destination}"
		)
	if modification.kind == "normalize-assbin-textures":
		if source_relative != destination_relative:
			raise ExperimentError(
				"Assbin normalization source and destination must be the same disposable file"
			)
		if not modification.texture_indices:
			raise ExperimentError("Assbin normalization requires texture_indices")
		if not destination.is_file():
			raise ExperimentError(f"Assbin normalization target is not a file: {destination}")
	return source, destination


def _safe_resource_relative_path(value: str) -> PurePosixPath:
	if "\\" in value:
		raise ExperimentError("Modification paths must use forward slashes")
	path = PurePosixPath(value)
	if path.is_absolute() or not path.parts or any(part in {"", ".", ".."} for part in path.parts):
		raise ExperimentError(f"Unsafe relative resource path: {value!r}")
	return path


def _prepare_disposable_game(
	config: AppConfig,
	result: ResultDirectory,
	logger: CommandLogger,
	modification: TestModification,
	disposable_game: Path,
	source: Path,
) -> dict[str, Any]:
	staging_root = (config.paths.app_root / _DISPOSABLE_DIRECTORY_NAME).absolute()
	_validate_disposable_path(staging_root, disposable_game, result.path.name)
	if disposable_game.exists() or disposable_game.is_symlink():
		raise ExperimentError(f"Disposable game path already exists: {disposable_game}")
	usage = shutil.disk_usage(staging_root.parent)
	copy_bytes = _allocated_tree_bytes(config.paths.live_game)
	if usage.free < copy_bytes + 2 * 1024**3:
		raise ExperimentError(
			f"Insufficient free space for disposable game copy: "
			f"need at least {copy_bytes + 2 * 1024**3}, have {usage.free}"
		)
	_require_command(
		logger.run(
			["mkdir", "-p", "--", str(staging_root)],
			timeout_seconds=15.0,
		),
		"create disposable staging directory",
	)
	_require_command(
		logger.run(
			[
				"cp",
				"-a",
				"--reflink=auto",
				"--sparse=always",
				f"{config.paths.live_game}/.",
				str(disposable_game),
			],
			timeout_seconds=900.0,
		),
		"copy disposable game tree",
	)
	destination_relative = _safe_resource_relative_path(modification.destination)
	destination = disposable_game.joinpath(*destination_relative.parts)
	if modification.kind == "copy-file":
		if destination.exists() or destination.is_symlink():
			raise ExperimentError(
				f"Disposable destination unexpectedly existed before copy: {destination}"
			)
		_require_command(
			logger.run(
				["cp", "--preserve=mode,timestamps", "--", str(source), str(destination)],
				timeout_seconds=30.0,
			),
			"copy selected resource",
		)
		normalization: dict[str, Any] | None = None
		destination_existed_before = False
	elif modification.kind == "normalize-assbin-textures":
		if not destination.is_file() or destination.is_symlink():
			raise ExperimentError(
				f"Disposable Assbin target is missing or unsafe: {destination}"
			)
		if sha256_file(destination) != sha256_file(source):
			raise ExperimentError("Disposable Assbin differs from live source before transform")
		temporary_output = destination.with_name(
			f".{destination.name}.{result.path.name}.normalized"
		)
		if temporary_output.exists() or temporary_output.is_symlink():
			raise ExperimentError(f"Normalization temporary output exists: {temporary_output}")
		normalize_result = logger.run(
			[
				sys.executable,
				"-m",
				"ptdiag",
				"scene",
				"normalize-textures",
				str(destination),
				str(temporary_output),
				*(str(index) for index in modification.texture_indices),
			],
			cwd=project_root(),
			timeout_seconds=240.0,
		)
		_require_command(normalize_result, "normalize embedded Assbin textures")
		try:
			normalization = json.loads(normalize_result.stdout)
		except json.JSONDecodeError as exc:
			raise ExperimentError("Normalizer returned invalid JSON") from exc
		if not isinstance(normalization, dict) or not normalization.get("ok"):
			raise ExperimentError("Normalizer did not report a verified result")
		_require_command(
			logger.run(
				["mv", "--", str(temporary_output), str(destination)],
				timeout_seconds=30.0,
			),
			"install normalized Assbin in disposable tree",
		)
		destination_existed_before = True
	else:
		raise ExperimentError(f"Unsupported modification kind: {modification.kind}")
	destination_hash = sha256_file(destination)
	source_hash = sha256_file(source)
	if modification.kind == "copy-file" and destination_hash != source_hash:
		raise ExperimentError("Copied resource hash differs from its source")
	if modification.kind == "normalize-assbin-textures" and destination_hash == source_hash:
		raise ExperimentError("Normalized Assbin hash did not change")
	ini_changes = _apply_disposable_ini_overrides(
		disposable_game,
		modification.ini_overrides,
		logger,
	)
	return {
		"staging_root": str(staging_root),
		"free_bytes_before": usage.free,
		"source_allocated_bytes": copy_bytes,
		"destination_path": str(destination),
		"destination_existed_before": destination_existed_before,
		"destination_sha256_after": destination_hash,
		"destination_size_after": destination.stat().st_size,
		"normalization": normalization,
		"ini_changes": ini_changes,
	}


def _inspect_ini_overrides(
	game_root: Path,
	overrides: Sequence[TestIniOverride],
) -> list[dict[str, Any]]:
	root = game_root.resolve(strict=True)
	result: list[dict[str, Any]] = []
	for override in overrides:
		path = _safe_game_file(root, override.path)
		state = inspect_ini_value(path, override.section, override.key)
		result.append(
			{
				"path": str(path),
				"section": override.section,
				"key": override.key,
				"expected": override.expected,
				"value": override.value,
				"current": state["value"],
				"matches_expected": state["value"] == override.expected,
				"matches_value": state["value"] == override.value,
				"sha256": state["sha256"],
				"size": state["size"],
				"mtime_ns": state["mtime_ns"],
			}
		)
	return result


def _apply_disposable_ini_overrides(
	disposable_game: Path,
	overrides: Sequence[TestIniOverride],
	logger: CommandLogger,
) -> list[dict[str, Any]]:
	root = disposable_game.resolve(strict=True)
	result: list[dict[str, Any]] = []
	for override in overrides:
		path = _safe_game_file(root, override.path)
		outcome = logger.run(
			_ini_override_command(path, override),
			cwd=project_root(),
			timeout_seconds=30.0,
		)
		_require_command(outcome, f"override [{override.section}] {override.key}")
		try:
			payload = json.loads(outcome.stdout)
		except json.JSONDecodeError as exc:
			raise ExperimentError("INI override returned invalid JSON") from exc
		if not isinstance(payload, dict) or not payload.get("ok"):
			raise ExperimentError("INI override did not report a verified result")
		result.append(payload)
	return result


def _safe_game_file(root: Path, relative: str) -> Path:
	parts = _safe_resource_relative_path(relative).parts
	path = root.joinpath(*parts)
	resolved = path.resolve(strict=True)
	try:
		resolved.relative_to(root)
	except ValueError as exc:
		raise ExperimentError(f"INI path escapes game tree: {relative}") from exc
	if path.is_symlink() or not resolved.is_file():
		raise ExperimentError(f"INI path is missing or unsafe: {path}")
	return resolved


def _ini_override_command(path: Path, override: TestIniOverride) -> list[str]:
	return [
		sys.executable,
		"-m",
		"ptdiag",
		"ini",
		"override",
		str(path),
		override.section,
		override.key,
		override.expected,
		override.value,
	]


def _replace_game_link(
	prefix_link: Path,
	disposable_game: Path,
	expected_live_game: Path,
	logger: CommandLogger,
) -> str:
	if not prefix_link.is_symlink():
		raise ExperimentError(f"Dedicated-prefix game path is not a symlink: {prefix_link}")
	if prefix_link.resolve(strict=True) != expected_live_game.resolve(strict=True):
		raise ExperimentError("Dedicated-prefix game link changed after source preflight")
	original_target = os.readlink(prefix_link)
	_require_command(
		logger.run(
			["ln", "-sfn", "--", str(disposable_game), str(prefix_link)],
			cwd=prefix_link.parent,
			timeout_seconds=15.0,
		),
		"point dedicated prefix at disposable game",
	)
	if not prefix_link.is_symlink() or prefix_link.resolve(strict=True) != disposable_game.resolve(strict=True):
		raise ExperimentError("Dedicated-prefix game link did not resolve to disposable copy")
	return original_target


def _restore_game_link(
	prefix_link: Path,
	original_target: str,
	expected_live_game: Path,
	logger: CommandLogger,
) -> None:
	_require_command(
		logger.run(
			["ln", "-sfn", "--", original_target, str(prefix_link)],
			cwd=prefix_link.parent,
			timeout_seconds=15.0,
		),
		"restore dedicated-prefix game link",
	)
	if not prefix_link.is_symlink():
		raise ExperimentError("Restored dedicated-prefix game path is not a symlink")
	if os.readlink(prefix_link) != original_target:
		raise ExperimentError("Dedicated-prefix game link text was not restored exactly")
	if prefix_link.resolve(strict=True) != expected_live_game.resolve(strict=True):
		raise ExperimentError("Dedicated-prefix game link target was not restored")


def _remove_disposable_game(
	config: AppConfig,
	result: ResultDirectory,
	disposable_game: Path,
	logger: CommandLogger,
) -> None:
	staging_root = (config.paths.app_root / _DISPOSABLE_DIRECTORY_NAME).absolute()
	_validate_disposable_path(staging_root, disposable_game, result.path.name)
	_require_command(
		logger.run(
			["rm", "-rf", "--one-file-system", "--", str(disposable_game)],
			timeout_seconds=900.0,
		),
		"remove disposable game tree",
	)
	if disposable_game.exists() or disposable_game.is_symlink():
		raise ExperimentError(f"Disposable game cleanup left its target: {disposable_game}")
	try:
		staging_root.rmdir()
	except OSError:
		pass


def _validate_disposable_path(staging_root: Path, path: Path, expected_name: str) -> None:
	if path.parent != staging_root or path.name != expected_name:
		raise ExperimentError(f"Unsafe disposable game path: {path}")
	if staging_root.name != _DISPOSABLE_DIRECTORY_NAME:
		raise ExperimentError(f"Unsafe disposable staging root: {staging_root}")


def _allocated_tree_bytes(root: Path) -> int:
	total = 0
	for directory, _, filenames in os.walk(root, followlinks=False):
		for filename in filenames:
			path = Path(directory) / filename
			try:
				total += path.lstat().st_blocks * 512
			except FileNotFoundError:
				continue
	return total


def _require_command(outcome: CommandResult, operation: str) -> None:
	if not _command_ok(outcome):
		raise ExperimentError(
			f"Could not {operation} ({outcome.returncode}): "
			f"{outcome.error or outcome.stderr.strip() or 'command failed'}"
		)


def _log_modification_plan(
	logger: CommandLogger,
	config: AppConfig,
	disposable_game: Path,
	source: Path,
	modification: TestModification,
) -> None:
	staging_root = disposable_game.parent
	destination = disposable_game.joinpath(
		*_safe_resource_relative_path(modification.destination).parts
	)
	prefix_link = config.proton7.prefix / "drive_c/Program Files (x86)/Pristontale EU"
	commands: list[tuple[list[str], float]] = [
		(["mkdir", "-p", "--", str(staging_root)], 15.0),
		(
			[
				"cp",
				"-a",
				"--reflink=auto",
				"--sparse=always",
				f"{config.paths.live_game}/.",
				str(disposable_game),
			],
			900.0,
		),
	]
	if modification.kind == "copy-file":
		commands.append(
			(["cp", "--preserve=mode,timestamps", "--", str(source), str(destination)], 30.0)
		)
	elif modification.kind == "normalize-assbin-textures":
		temporary_output = destination.with_name(f".{destination.name}.planned.normalized")
		commands.extend(
			(
				(
					[
						sys.executable,
						"-m",
						"ptdiag",
						"scene",
						"normalize-textures",
						str(destination),
						str(temporary_output),
						*(str(index) for index in modification.texture_indices),
					],
					240.0,
				),
				(["mv", "--", str(temporary_output), str(destination)], 30.0),
			)
		)
	else:
		raise ExperimentError(f"Unsupported modification kind: {modification.kind}")
	for override in modification.ini_overrides:
		ini_path = disposable_game.joinpath(
			*_safe_resource_relative_path(override.path).parts
		)
		commands.append((_ini_override_command(ini_path, override), 30.0))
	commands.extend(
		(
			(["ln", "-sfn", "--", str(disposable_game), str(prefix_link)], 15.0),
			(["ln", "-sfn", "--", os.readlink(prefix_link), str(prefix_link)], 15.0),
			(["rm", "-rf", "--one-file-system", "--", str(disposable_game)], 900.0),
		)
	)
	for command, timeout in commands:
		logger.run(command, timeout_seconds=timeout, dry_run=True)


def _setup_failure_summary(
	testcase: Testcase,
	result: ResultDirectory,
	dry_run: bool,
	errors: Sequence[str],
	modification: Mapping[str, Any],
) -> dict[str, Any]:
	return {
		"ok": False,
		"status": "preflight-failed" if errors else "setup-error",
		"dry_run": dry_run,
		"test": testcase.name,
		"result_directory": str(result.path),
		"preflight": {"ok": False, "errors": list(errors)},
		"modification": dict(modification),
		"native_log": {"source": None, "current_run_bytes": 0},
		"registry": {"restored": None},
	}


def _preflight(
	config: AppConfig,
	testcase: Testcase,
	environment: Mapping[str, Any],
	hashes: list[dict[str, Any]],
) -> list[str]:
	errors: list[str] = []
	if not environment.get("ok"):
		errors.append("Required environment checks failed")
	if any(not item.get("ok", False) for item in hashes if item.get("required", False)):
		errors.append("A required baseline hash is missing or mismatched")
	if testcase.runner != "proton7":
		errors.append(f"Unsupported runner: {testcase.runner}")
	if testcase.game.args != config.game.args:
		errors.append("Testcase game arguments differ from the known PTEU_REBORN baseline")
	if testcase.game.cwd != "live_game":
		errors.append("Only the verified live_game working directory is supported")
	if testcase.display.size != config.game.virtual_desktop:
		errors.append("Testcase desktop size differs from the 1600x900 baseline")
	if testcase.capture.strace_files and shutil.which("strace") is None:
		errors.append("scene-strace requires strace")
	if not testcase.watch.require_new_native_log:
		errors.append("Runtime tests must require a current-run native log")
	try:
		prefix = validate_dedicated_prefix(config)
	except (OSError, ValueError, PrefixSafetyError) as exc:
		errors.append(str(exc))
	else:
		linked_game = prefix / "drive_c/Program Files (x86)/Pristontale EU"
		try:
			if linked_game.resolve(strict=True) != config.paths.live_game.resolve(strict=True):
				errors.append("Dedicated prefix game link does not resolve to live_game")
		except OSError as exc:
			errors.append(f"Dedicated prefix game link is invalid: {exc}")
	active = _game_processes()
	if active:
		errors.append(f"Game.exe is already running: {active}")
	return errors


def _dry_run_commands(
	config: AppConfig,
	testcase: Testcase,
	logger: CommandLogger,
	registry_before: DesktopRegistrySnapshot,
	launch_argv: list[str],
	env_overrides: Mapping[str, str],
) -> None:
	cleanup_dedicated_wineserver(config, logger, timeout_seconds=15.0, dry_run=True)
	if testcase.display.virtual_desktop:
		for command in _registry_setup_commands(
			config, registry_before.desktop_name, testcase.display.size
		):
			logger.run(
				command,
				cwd=config.paths.live_game,
				env_overrides=_proton_environment(config),
				timeout_seconds=15.0,
				dry_run=True,
			)
	logger.run(
		launch_argv,
		cwd=config.paths.live_game,
		env_overrides=env_overrides,
		timeout_seconds=testcase.timeout_seconds,
		dry_run=True,
	)
	cleanup_dedicated_wineserver(config, logger, timeout_seconds=15.0, dry_run=True)
	if testcase.display.virtual_desktop:
		for command in _registry_restore_commands(config, registry_before):
			logger.run(
				command,
				cwd=config.paths.live_game,
				env_overrides=_proton_environment(config),
				timeout_seconds=15.0,
				dry_run=True,
			)


def _run_and_watch(
	config: AppConfig,
	testcase: Testcase,
	result: ResultDirectory,
	logger: CommandLogger,
	argv: Sequence[str],
	env_overrides: Mapping[str, str],
	baseline: Mapping[str, LogFileState],
) -> dict[str, Any]:
	runner_log_path = result.path / "runner.log"
	environment = os.environ.copy()
	environment.update(env_overrides)
	test_start_ns = time.time_ns()
	started_at = datetime.now().astimezone()
	write_json(
		result.path / "test-start.json",
		{
			"test_start_ns": test_start_ns,
			"test_start_epoch": test_start_ns / 1_000_000_000,
			"recorded_at": started_at.isoformat(),
		},
	)
	process: subprocess.Popen[str] | None = None
	process_identity: dict[str, Any] | None = None
	failure_state: LogFileState | None = None
	failure_trigger_mtime_ns: int | None = None
	current_state: LogFileState | None = None
	current_analysis: dict[str, Any] | None = None
	current_text = ""
	current_log_captures: list[dict[str, Any]] = []
	performance_sampler: PerformanceSampler | None = None
	performance_summary: dict[str, Any] | None = None
	detected_at_ns: int | None = None
	timed_out = False
	stop_reason = "process-exited"
	with runner_log_path.open("a", encoding="utf-8") as runner_log:
		try:
			process = subprocess.Popen(
				list(argv),
				cwd=config.paths.live_game,
				env=environment,
				stdout=runner_log,
				stderr=subprocess.STDOUT,
				text=True,
				start_new_session=True,
			)
		except OSError as exc:
			logger.record_lifecycle(
				argv,
				state="launch-error",
				cwd=config.paths.live_game,
				env_overrides=env_overrides,
				timeout_seconds=testcase.timeout_seconds,
				started_at=started_at,
				error=str(exc),
			)
			raise ExperimentError(f"Could not launch experiment: {exc}") from exc
		assert process is not None
		logger.record_lifecycle(
			argv,
			state="started",
			cwd=config.paths.live_game,
			env_overrides=env_overrides,
			timeout_seconds=testcase.timeout_seconds,
			started_at=started_at,
			pid=process.pid,
		)
		deadline_ns = test_start_ns + int(testcase.timeout_seconds * 1_000_000_000)
		post_failure_deadline_ns: int | None = None
		while process.poll() is None:
			now_ns = time.time_ns()
			if process_identity is None:
				process_identity = _capture_game_process(testcase, result.path)
				if process_identity is not None and testcase.capture.performance_stats:
					try:
						performance_sampler = PerformanceSampler(
							int(process_identity["pid"]),
							result.path / "performance.csv",
						)
					except (OSError, ValueError, KeyError) as exc:
						performance_summary = {"error": str(exc)}
			if performance_sampler is not None:
				performance_sampler.maybe_sample(now_ns)
			if failure_state is None:
				for state in discover_current_native_logs(
					config.paths.native_logs, baseline, test_start_ns
				):
					analysis = parse_current_run_log(state, baseline)
					current_state = state
					current_analysis = analysis.to_summary()
					current_text = read_current_run_log_text(state, baseline)
					if analysis.load_texture_to_material_failed:
						failure_state = state
						failure_trigger_mtime_ns = state.mtime_ns
						detected_at_ns = now_ns
						post_failure_deadline_ns = now_ns + int(
							testcase.stop_after_failure_seconds * 1_000_000_000
						)
						break
			if post_failure_deadline_ns is not None and now_ns >= post_failure_deadline_ns:
				stop_reason = "load-texture-to-material"
				_stop_process_group(process)
				break
			if now_ns >= deadline_ns:
				timed_out = True
				stop_reason = "timeout"
				_stop_process_group(process)
				break
			time.sleep(0.1)
		if process.poll() is None:
			_stop_process_group(process)
		returncode = process.wait(timeout=5.0)
		if performance_sampler is not None:
			performance_summary = performance_sampler.close()
		logger.record_lifecycle(
			argv,
			state="finished",
			cwd=config.paths.live_game,
			env_overrides=env_overrides,
			timeout_seconds=testcase.timeout_seconds,
			started_at=started_at,
			pid=process.pid,
			returncode=returncode,
			timed_out=timed_out,
		)

	final_current = discover_current_native_logs(
		config.paths.native_logs, baseline, test_start_ns
	)
	if final_current:
		candidates: list[
			tuple[tuple[int, int, int, int, str], LogFileState, str, dict[str, Any]]
		] = []
		for index, state in enumerate(final_current):
			text = read_current_run_log_text(state, baseline)
			analysis = parse_current_run_log(state, baseline).to_summary()
			scenes = analysis.get("scenes")
			scene_names = (
				sorted(str(name) for name in scenes)
				if isinstance(scenes, Mapping)
				else []
			)
			captured_path = (
				result.path / "native-logs" / f"{index:02d}-{state.path.name}.current.log"
			)
			_write_text(captured_path, text)
			current_log_captures.append(
				{
					"source": str(state.path),
					"state": _serialize_log_state(state),
					"captured_path": str(captured_path),
					"current_run_bytes": len(text.encode("utf-8")),
					"scene_names": scene_names,
					"load_texture_to_material_failed": analysis.get(
						"load_texture_to_material_failed", False
					),
					"import_scene_threw": analysis.get("import_scene_threw", False),
				}
			)
			candidates.append(
				(_native_log_priority(state, analysis), state, text, analysis)
			)
		_, current_state, current_text, current_analysis = max(
			candidates,
			key=lambda item: item[0],
		)
	if failure_state is not None:
		final_states = snapshot_native_logs(config.paths.native_logs)
		final_state = final_states.get(str(failure_state.path.absolute()))
		if final_state is not None:
			current_state = final_state
			current_text = read_current_run_log_text(final_state, baseline)
			current_analysis = parse_current_run_log(final_state, baseline).to_summary()
	if current_state is not None:
		_write_text(result.path / "native.log", current_text)
	whispering_vale = (
		current_analysis.get("whispering_vale")
		if isinstance(current_analysis, Mapping)
		else None
	)
	scene_import_survived = bool(
		whispering_vale
		and not timed_out
		and not current_analysis.get("load_texture_to_material_failed", False)
		and not current_analysis.get("import_scene_threw", False)
		and not current_analysis.get("failed_import_scenes", [])
	) if isinstance(current_analysis, Mapping) else False
	trigger_ns = (
		max(test_start_ns, failure_trigger_mtime_ns)
		if failure_trigger_mtime_ns is not None
		else None
	)
	return {
		"test_start_ns": test_start_ns,
		"test_start_epoch": test_start_ns / 1_000_000_000,
		"finished_at": datetime.now().astimezone().isoformat(),
		"launcher_pid": process.pid if process is not None else None,
		"game_process": process_identity,
		"returncode": process.returncode if process is not None else None,
		"timed_out": timed_out,
		"stop_reason": stop_reason,
		"failure_detected": failure_state is not None,
		"scene_import_survived": scene_import_survived,
		"detected_at_ns": detected_at_ns,
		"trigger_epoch": trigger_ns / 1_000_000_000 if trigger_ns is not None else None,
		"analysis": current_analysis,
		"native_logs": current_log_captures,
		"performance": performance_summary,
		"native_log": {
			"source": str(current_state.path) if current_state else None,
			"state": _serialize_log_state(current_state) if current_state else None,
			"current_run_bytes": len(current_text.encode("utf-8")),
		},
	}


def _native_log_priority(
	state: LogFileState,
	analysis: Mapping[str, Any],
) -> tuple[int, int, int, int, str]:
	scenes = analysis.get("scenes")
	scene_count = 0
	if isinstance(scenes, Mapping):
		for imports in scenes.values():
			if isinstance(imports, list):
				scene_count += len(imports)
	return (
		int(scene_count > 0),
		scene_count,
		int(state.path.name.casefold().startswith("log_")),
		state.mtime_ns,
		str(state.path),
	)


def _stop_process_group(process: subprocess.Popen[str]) -> None:
	if process.poll() is not None:
		return
	for signum, timeout in (
		(signal.SIGINT, 5.0),
		(signal.SIGTERM, 5.0),
		(signal.SIGKILL, 3.0),
	):
		try:
			os.killpg(process.pid, signum)
		except ProcessLookupError:
			return
		try:
			process.wait(timeout=timeout)
			return
		except subprocess.TimeoutExpired:
			continue


def _capture_game_process(testcase: Testcase, result_directory: Path) -> dict[str, Any] | None:
	processes = _game_processes()
	if not processes:
		return None
	pid = processes[0]
	root = Path("/proc") / str(pid)
	identity: dict[str, Any] = {"pid": pid}
	try:
		identity["comm"] = (root / "comm").read_text(encoding="utf-8").strip()
		identity["cmdline"] = (
			(root / "cmdline").read_bytes().decode("utf-8", errors="replace").split("\0")
		)
		identity["cwd"] = str((root / "cwd").readlink())
		identity["exe"] = str((root / "exe").readlink())
		if testcase.capture.process_maps:
			shutil.copyfile(root / "maps", result_directory / "process-maps.txt")
	except OSError as exc:
		identity["capture_error"] = str(exc)
	return identity


def _game_processes() -> list[int]:
	result: list[int] = []
	for entry in Path("/proc").iterdir():
		if not entry.name.isdigit():
			continue
		try:
			comm = (entry / "comm").read_text(encoding="utf-8").strip().casefold()
		except OSError:
			continue
		if comm in {"game.exe", "gamese.exe"}:
			result.append(int(entry.name))
	return sorted(result)


def _launch_argv(config: AppConfig, testcase: Testcase, result_directory: Path) -> list[str]:
	game = f"{_WINDOWS_GAME_DIRECTORY}\\{config.game.exe}"
	proton = _proton_command(config, "runinprefix", game, *testcase.game.args)
	if not testcase.capture.strace_files:
		return proton
	raw_directory = result_directory / "raw-strace"
	raw_directory.mkdir(exist_ok=False)
	strace = shutil.which("strace") or "/usr/bin/strace"
	return [
		strace,
		"-ff",
		"-qq",
		"-ttt",
		"-s",
		"1024",
		"-e",
		"trace=%file",
		"-o",
		str(raw_directory / "trace"),
		*proton,
	]


def _proton_command(config: AppConfig, *arguments: str) -> list[str]:
	shim = project_root() / "ptlinux/proton_compat.py"
	return [
		sys.executable,
		str(shim),
		str(config.proton7.proton),
		*arguments,
	]


def _runtime_environment(
	config: AppConfig,
	testcase: Testcase,
	result_directory: Path,
) -> dict[str, str]:
	environment = _proton_environment(config)
	environment.update(testcase.env)
	if environment.get("DXVK_STATE_CACHE_PATH") == _PERSISTENT_STATE_CACHE_SENTINEL:
		environment["DXVK_STATE_CACHE_PATH"] = str(
			(config.paths.app_root / _PERSISTENT_STATE_CACHE_DIRECTORY).absolute()
		)
	environment.update(
		{
			"DXVK_LOG_PATH": str(result_directory),
			"PROTON_LOG": "0",
			"STEAM_COMPAT_INSTALL_PATH": str(config.paths.live_game),
			"SteamAppId": "0",
			"SteamGameId": "0",
		}
	)
	return environment


def _state_cache_plan(
	config: AppConfig,
	environment: Mapping[str, str],
) -> dict[str, Any] | None:
	configured = environment.get("DXVK_STATE_CACHE_PATH")
	if configured is None:
		return None
	cache_directory = Path(configured).expanduser().absolute()
	cache_file = cache_directory / f"{Path(config.game.exe).stem}.dxvk-cache"
	return {
		"directory": str(cache_directory),
		"cache_file": str(cache_file),
		"persistent": True,
		"seed_source": str(config.paths.live_game / cache_file.name),
		"seeded": False,
		"before": _state_cache_file_snapshot(cache_file),
		"after": None,
	}


def _prepare_state_cache(
	config: AppConfig,
	environment: Mapping[str, str],
	logger: CommandLogger,
) -> dict[str, Any]:
	plan = _state_cache_plan(config, environment)
	if plan is None:
		raise ExperimentError("Persistent DXVK state-cache plan is missing")
	cache_directory = Path(str(plan["directory"]))
	expected = (config.paths.app_root / _PERSISTENT_STATE_CACHE_DIRECTORY).absolute()
	if cache_directory != expected:
		raise ExperimentError(
			f"Refusing unmanaged DXVK state-cache directory: {cache_directory}; "
			f"expected {expected}"
		)
	_require_command(
		logger.run(
			["mkdir", "-p", "--", str(cache_directory)],
			timeout_seconds=15.0,
		),
		"create managed DXVK state-cache directory",
	)
	cache_file = Path(str(plan["cache_file"]))
	seed_source = Path(str(plan["seed_source"]))
	if not cache_file.exists() and seed_source.is_file():
		_require_command(
			logger.run(
				[
					"cp",
					"--preserve=mode,timestamps",
					"--",
					str(seed_source),
					str(cache_file),
				],
				timeout_seconds=30.0,
			),
			"seed managed DXVK state cache",
		)
		plan["seeded"] = True
	plan["before"] = _state_cache_file_snapshot(cache_file)
	return plan


def _state_cache_file_snapshot(path: Path) -> dict[str, Any]:
	result: dict[str, Any] = {
		"path": str(path),
		"exists": path.is_file(),
		"size": None,
		"mtime_ns": None,
		"sha256": None,
	}
	if not result["exists"]:
		return result
	try:
		state = path.stat()
		result.update(
			{
				"size": state.st_size,
				"mtime_ns": state.st_mtime_ns,
				"sha256": sha256_file(path),
			}
		)
	except OSError as exc:
		result["error"] = str(exc)
	return result


def _proton_environment(config: AppConfig) -> dict[str, str]:
	return {
		"STEAM_COMPAT_CLIENT_INSTALL_PATH": str(config.proton7.steam_root),
		"STEAM_COMPAT_DATA_PATH": str(config.proton7.compat),
		"WINEPREFIX": str(config.proton7.prefix),
		"WINEDEBUG": "-all",
	}


def _registry_setup_commands(
	config: AppConfig,
	desktop_name: str,
	desktop_size: str,
) -> list[list[str]]:
	return [
		_proton_command(
			config,
			"runinprefix",
			"reg.exe",
			"add",
			_EXPLORER_KEY,
			"/v",
			"Desktop",
			"/t",
			"REG_SZ",
			"/d",
			desktop_name,
			"/f",
		),
		_proton_command(
			config,
			"runinprefix",
			"reg.exe",
			"add",
			_DESKTOPS_KEY,
			"/v",
			desktop_name,
			"/t",
			"REG_SZ",
			"/d",
			desktop_size,
			"/f",
		),
	]


def _registry_restore_commands(
	config: AppConfig,
	snapshot: DesktopRegistrySnapshot,
) -> list[list[str]]:
	return [
		_registry_restore_command(
			config,
			_EXPLORER_KEY,
			"Desktop",
			snapshot.selected_desktop.exists,
			snapshot.selected_desktop.value,
		),
		_registry_restore_command(
			config,
			_DESKTOPS_KEY,
			snapshot.desktop_name,
			snapshot.desktop_size.exists,
			snapshot.desktop_size.value,
		),
	]


def _registry_restore_command(
	config: AppConfig,
	key: str,
	name: str,
	existed: bool,
	value: str | None,
) -> list[str]:
	if existed:
		assert value is not None
		return _proton_command(
			config,
			"runinprefix",
			"reg.exe",
			"add",
			key,
			"/v",
			name,
			"/t",
			"REG_SZ",
			"/d",
			value,
			"/f",
		)
	return _proton_command(
		config,
		"runinprefix",
		"reg.exe",
		"delete",
		key,
		"/v",
		name,
		"/f",
	)


def _run_checked_commands(
	logger: CommandLogger,
	commands: Sequence[Sequence[str]],
	*,
	cwd: Path,
	env_overrides: Mapping[str, str],
) -> None:
	for command in commands:
		outcome = logger.run(
			command,
			cwd=cwd,
			env_overrides=env_overrides,
			timeout_seconds=15.0,
		)
		if not _command_ok(outcome):
			raise ExperimentError(
				f"Command failed ({outcome.returncode}): {list(command)!r}: "
				f"{outcome.error or outcome.stderr.strip()}"
			)


def _cleanup_checked(config: AppConfig, logger: CommandLogger) -> None:
	outcome = cleanup_dedicated_wineserver(
		config,
		logger,
		timeout_seconds=15.0,
	)
	if (
		outcome.returncode == 1
		and not outcome.timed_out
		and outcome.error is None
		and not outcome.stdout
		and not outcome.stderr
	):
		# Proton 7 wineserver -k returns 1 when this dedicated prefix has no
		# server to stop.  That is the desired idempotent cleanup state.
		return
	if not _command_ok(outcome):
		raise ExperimentError(
			"Dedicated-prefix wineserver cleanup failed: "
			f"{outcome.error or outcome.stderr.strip() or outcome.returncode}"
		)


def _command_ok(result: CommandResult) -> bool:
	return not result.timed_out and result.error is None and result.returncode == 0


def _collect_hashes(config: AppConfig, testcase: Testcase) -> list[dict[str, Any]]:
	items: list[dict[str, Any]] = []
	for verification in verify_known_game_hashes(config.paths.live_game):
		item = verification.to_dict()
		item["required"] = True
		item["role"] = "known-game-baseline"
		items.append(item)
	extra = {
		config.proton7.proton: "proton-launcher",
		config.proton7.prefix / "drive_c/windows/syswow64/d3dx9_43.dll": "native-d3dx9-43",
		config.proton7.prefix
		/ "drive_c/windows/syswow64/d3dcompiler_43.dll": "native-d3dcompiler-43",
		config.proton7.prefix / "drive_c/windows/syswow64/d3d9.dll": "dxvk-d3d9",
		project_root() / "ptlinux/proton_compat.py": "proton-python314-shim",
		config.source: "local-config",
		testcase.source: "testcase",
	}
	dxvk_config = testcase.env.get("DXVK_CONFIG_FILE")
	if dxvk_config:
		dxvk_config_path = Path(dxvk_config).expanduser()
		if not dxvk_config_path.is_absolute():
			dxvk_config_path = (project_root() / dxvk_config_path).absolute()
		extra[dxvk_config_path] = "dxvk-config"
	for path, role in extra.items():
		item: dict[str, Any] = {
			"path": str(path),
			"role": role,
			"required": role not in {"local-config", "testcase"},
			"expected_sha256": None,
			"actual_sha256": None,
			"ok": False,
			"error": None,
		}
		try:
			item["actual_sha256"] = sha256_file(path)
			item["ok"] = True
		except OSError as exc:
			item["error"] = str(exc)
		items.append(item)
	return items


def _collect_dll_overrides(prefix: Path) -> dict[str, str]:
	sections = read_registry_string_values(prefix / "user.reg")
	return dict(sorted(sections.get(r"Software\Wine\DllOverrides", {}).items()))


def _serialize_log_states(states: Mapping[str, LogFileState]) -> list[dict[str, Any]]:
	return [_serialize_log_state(state) for state in sorted(states.values(), key=lambda x: str(x.path))]


def _serialize_log_state(state: LogFileState) -> dict[str, Any]:
	return {
		"path": str(state.path),
		"mtime_ns": state.mtime_ns,
		"size": state.size,
		"inode": state.inode,
	}


def _desktop_name(result_directory: Path) -> str:
	digits = "".join(character for character in result_directory.name if character.isdigit())
	return f"PTDiag{digits[-12:]}"


def _finalize(result: ResultDirectory, summary: dict[str, Any]) -> None:
	write_json(result.path / "summary.json", summary)
	_write_text(result.path / "report.md", _report(summary))


def _report(summary: Mapping[str, Any]) -> str:
	preflight = summary.get("preflight", {})
	launch = summary.get("launch", {})
	registry = summary.get("registry", {})
	native_log = summary.get("native_log", {})
	modification = summary.get("modification")
	state_cache = summary.get("dxvk_state_cache")
	runtime = summary.get("runtime")
	lines = [
		f"# {summary.get('test', 'experiment')}",
		"",
		f"Status: `{summary.get('status')}`",
		f"Result directory: `{summary.get('result_directory')}`",
		f"Dry run: `{summary.get('dry_run')}`",
		"",
		"## Preflight",
		"",
		f"Passed: `{preflight.get('ok')}`" if isinstance(preflight, Mapping) else "Passed: `False`",
	]
	if isinstance(preflight, Mapping):
		for error in preflight.get("errors", []):
			lines.append(f"- {error}")
	lines.extend(
		[
			"",
			"## Launch",
			"",
			f"CWD: `{launch.get('cwd')}`" if isinstance(launch, Mapping) else "",
			f"Timeout: `{launch.get('timeout_seconds')}` seconds"
			if isinstance(launch, Mapping)
			else "",
			"",
			"## Current-run native log",
			"",
			f"Source: `{native_log.get('source')}`" if isinstance(native_log, Mapping) else "",
			f"Captured bytes: `{native_log.get('current_run_bytes')}`"
			if isinstance(native_log, Mapping)
			else "",
			"",
			"## Registry rollback",
			"",
			f"Restored: `{registry.get('restored')}`" if isinstance(registry, Mapping) else "",
		]
	)
	if summary.get("review_gate"):
		lines.extend(["", "## Review gate", "", str(summary["review_gate"])])
	if isinstance(runtime, Mapping) and isinstance(runtime.get("performance"), Mapping):
		performance = runtime["performance"]
		lines.extend(
			[
				"",
				"## Performance capture",
				"",
				f"Samples: `{performance.get('sample_count')}`",
				f"Interval: `{performance.get('interval_seconds')}` seconds",
				f"CSV: `{performance.get('output')}`",
			]
		)
	if isinstance(state_cache, Mapping):
		lines.extend(
			[
				"",
				"## DXVK state cache",
				"",
				f"Directory: `{state_cache.get('directory')}`",
				f"Seeded: `{state_cache.get('seeded')}`",
				f"Before: `{state_cache.get('before')}`",
				f"After: `{state_cache.get('after')}`",
			]
		)
	if isinstance(modification, Mapping):
		lines.extend(
			[
				"",
				"## Disposable modification",
				"",
				f"Kind: `{modification.get('kind')}`",
				f"Source: `{modification.get('source')}`",
				f"Destination: `{modification.get('destination')}`",
				f"Prepared: `{modification.get('prepared')}`",
				f"Prefix link restored: `{modification.get('prefix_link_restored')}`",
				f"Disposable copy removed: `{modification.get('disposable_removed')}`",
			]
		)
	if summary.get("error"):
		lines.extend(["", "## Error", "", str(summary["error"])])
	return "\n".join(line for line in lines if line is not None).rstrip() + "\n"


def _write_text(path: Path, value: str) -> None:
	path.parent.mkdir(parents=True, exist_ok=True)
	temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
	temporary.write_text(value.rstrip("\n") + "\n", encoding="utf-8")
	temporary.replace(path)