pristontale-eu-linux/src/ptdiag/experiments/log_selection.py

"""Current-run native-log observation, capture, and selection."""

from __future__ import annotations

from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Protocol

from ..logs import (
	LogFileState,
	discover_current_native_logs,
	parse_current_run_log,
	read_current_run_log_text,
	snapshot_native_logs,
)


class NativeLogAnalysis(Protocol):
	load_texture_to_material_failed: bool
	asset_pack_health_failed: bool

	def to_summary(self) -> dict[str, Any]: ...


NativeLogDiscoverer = Callable[
	[Path, Mapping[str, LogFileState], int],
	Sequence[LogFileState],
]
NativeLogParser = Callable[
	[LogFileState, Mapping[str, LogFileState]],
	NativeLogAnalysis,
]
NativeLogReader = Callable[
	[LogFileState, Mapping[str, LogFileState]],
	str,
]
NativeLogSnapshotter = Callable[[Path], Mapping[str, LogFileState]]
LogStateSerializer = Callable[[LogFileState], dict[str, Any]]
NativeLogPrioritizer = Callable[
	[LogFileState, Mapping[str, Any]],
	tuple[int, int, int, int, str],
]
TextWriter = Callable[[Path, str], None]


@dataclass(frozen=True)
class NativeLogObservation:
	state: LogFileState
	text: str
	analysis: dict[str, Any]
	failure_reason: str | None


@dataclass(frozen=True)
class NativeLogSelection:
	observation: NativeLogObservation | None
	captures: tuple[dict[str, Any], ...]


def observe_current_native_logs(
	log_directory: Path,
	baseline: Mapping[str, LogFileState],
	test_start_ns: int,
	*,
	discoverer: NativeLogDiscoverer = discover_current_native_logs,
	parser: NativeLogParser = parse_current_run_log,
	reader: NativeLogReader = read_current_run_log_text,
) -> tuple[NativeLogObservation, ...]:
	"""Parse each current-run candidate and identify an actionable failure."""

	observations: list[NativeLogObservation] = []
	for state in discoverer(log_directory, baseline, test_start_ns):
		parsed = parser(state, baseline)
		failure_reason = (
			"asset-pack-health"
			if parsed.asset_pack_health_failed
			else "load-texture-to-material"
			if parsed.load_texture_to_material_failed
			else None
		)
		observations.append(
			NativeLogObservation(
				state=state,
				text=reader(state, baseline),
				analysis=parsed.to_summary(),
				failure_reason=failure_reason,
			)
		)
	return tuple(observations)


def capture_current_native_logs(
	log_directory: Path,
	baseline: Mapping[str, LogFileState],
	test_start_ns: int,
	result_directory: Path,
	*,
	fallback: NativeLogObservation | None,
	failure_state: LogFileState | None,
	serialize_state: LogStateSerializer,
	write_text: TextWriter,
	prioritize: NativeLogPrioritizer | None = None,
	discoverer: NativeLogDiscoverer = discover_current_native_logs,
	parser: NativeLogParser = parse_current_run_log,
	reader: NativeLogReader = read_current_run_log_text,
	snapshotter: NativeLogSnapshotter = snapshot_native_logs,
) -> NativeLogSelection:
	"""Capture all current-run logs and choose the strongest primary evidence."""

	captures: list[dict[str, Any]] = []
	candidates: list[NativeLogObservation] = []
	for index, state in enumerate(discoverer(log_directory, baseline, test_start_ns)):
		text = reader(state, baseline)
		analysis = parser(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_directory
			/ "native-logs"
			/ f"{index:02d}-{state.path.name}.current.log"
		)
		write_text(captured_path, text)
		captures.append(
			{
				"source": str(state.path),
				"state": serialize_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,
				),
				"asset_pack_health_failed": analysis.get(
					"asset_pack_health_failed",
					False,
				),
				"import_scene_threw": analysis.get("import_scene_threw", False),
			}
		)
		candidates.append(
			NativeLogObservation(
				state=state,
				text=text,
				analysis=analysis,
				failure_reason=None,
			)
		)
	prioritizer = native_log_priority if prioritize is None else prioritize
	selected = (
		max(candidates, key=lambda item: prioritizer(item.state, item.analysis))
		if candidates
		else fallback
	)
	if failure_state is not None:
		final_states = snapshotter(log_directory)
		final_state = final_states.get(str(failure_state.path.absolute()))
		if final_state is not None:
			selected = NativeLogObservation(
				state=final_state,
				text=reader(final_state, baseline),
				analysis=parser(final_state, baseline).to_summary(),
				failure_reason=None,
			)
	if selected is not None:
		write_text(result_directory / "native.log", selected.text)
	return NativeLogSelection(observation=selected, captures=tuple(captures))


def native_log_priority(
	state: LogFileState,
	analysis: Mapping[str, Any],
) -> tuple[int, int, int, int, str]:
	"""Rank scene-rich engine logs ahead of newer low-signal logs."""

	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 scene_import_survived(
	analysis: Mapping[str, Any] | None,
	*,
	timed_out: bool,
) -> bool:
	"""Return whether the target scene completed without a captured failure."""

	if analysis is None:
		return False
	return bool(
		analysis.get("whispering_vale")
		and not timed_out
		and not analysis.get("load_texture_to_material_failed", False)
		and not analysis.get("asset_pack_health_failed", False)
		and not analysis.get("import_scene_threw", False)
		and not analysis.get("failed_import_scenes", [])
	)