9 Commits

22 changed files with 2493 additions and 272 deletions

View File

@@ -1,6 +1,6 @@
# PristonTale EU for Linux
Version 0.9.0 installs, updates, repairs, and launches PristonTale EU through a
Version 0.9.5 installs, updates, repairs, and launches PristonTale EU through a
native Linux dashboard. Game files come directly from the publisher and are not
included in this archive.
@@ -15,11 +15,11 @@ Then run:
```text
openssl pkey -pubin -in launcher-update-public.pem -outform DER | sha256sum
openssl dgst -sha256 -verify launcher-update-public.pem \
-signature pristontale-eu-linux-installer-v0-9-0.tar.gz.sig \
pristontale-eu-linux-installer-v0-9-0.tar.gz
sha256sum -c pristontale-eu-linux-installer-v0-9-0.tar.gz.sha256
tar -xzf pristontale-eu-linux-installer-v0-9-0.tar.gz
cd pristontale-eu-linux-installer-v0-9-0
-signature pristontale-eu-linux-installer-v0-9-5.tar.gz.sig \
pristontale-eu-linux-installer-v0-9-5.tar.gz
sha256sum -c pristontale-eu-linux-installer-v0-9-5.tar.gz.sha256
tar -xzf pristontale-eu-linux-installer-v0-9-5.tar.gz
cd pristontale-eu-linux-installer-v0-9-5
./install.sh
```
@@ -88,11 +88,3 @@ pristontale-eu cache-compatibility PATH Cache a verified compatibility pai
Use `./install.sh --dry-run` to preview dependency installation or
`./install.sh --skip-deps` to manage distribution packages and the host sandbox
manually.
On Ubuntu, install and launcher-update operations run a harmless Bubblewrap
user-namespace probe. If AppArmor produces the exact bwrap UID-map permission
denial, the launcher can install and load a pinned upstream restrictive bwrap
profile. It also repairs only UMU's known broken
`bwrap-userns-restrict-umu` link. Authentication is required; the repair does
not disable AppArmor or change global kernel security settings, and it refuses
to install a competing profile when another bwrap policy is present.

18
install.sh Executable file → Normal file
View File

@@ -155,6 +155,13 @@ if [[ ! -d "${SOURCE_ROOT}/ptlinux" \
|| ! -f "${SOURCE_ROOT}/ptlinux/ui/dashboard.py" \
|| ! -f "${SOURCE_ROOT}/ptlinux/launcher-update-public.pem" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/pt-identity-probe.exe" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/gamescope-3.16.20-ubuntu26.04-amd64" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/gamescope-compat.README" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/gamescope-compat.copyright" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/gamescope-lazy-output-images.patch" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/gamescope-wayland-live-resize.patch" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/gamescope-wayland-letterbox-backing.patch" \
|| ! -f "${SOURCE_ROOT}/ptlinux/native/gamescope-wayland-libdecor-input-scope.patch" \
|| ! -f "${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.desktop" \
|| ! -f "${SOURCE_ROOT}/packaging/eu.pristontale.PTLinux.png" \
|| ! -f "${SOURCE_ROOT}/packaging/pristontale-eu" \
@@ -237,6 +244,17 @@ install -m 0644 \
install -m 0644 \
"${SOURCE_ROOT}/ptlinux/native/pt-identity-probe.exe" \
"${STAGING_RELEASE}/ptlinux/native/pt-identity-probe.exe"
install -m 0755 \
"${SOURCE_ROOT}/ptlinux/native/gamescope-3.16.20-ubuntu26.04-amd64" \
"${STAGING_RELEASE}/ptlinux/native/gamescope-3.16.20-ubuntu26.04-amd64"
install -m 0644 \
"${SOURCE_ROOT}/ptlinux/native/gamescope-compat.README" \
"${SOURCE_ROOT}/ptlinux/native/gamescope-compat.copyright" \
"${SOURCE_ROOT}/ptlinux/native/gamescope-lazy-output-images.patch" \
"${SOURCE_ROOT}/ptlinux/native/gamescope-wayland-live-resize.patch" \
"${SOURCE_ROOT}/ptlinux/native/gamescope-wayland-letterbox-backing.patch" \
"${SOURCE_ROOT}/ptlinux/native/gamescope-wayland-libdecor-input-scope.patch" \
"${STAGING_RELEASE}/ptlinux/native/"
install -m 0644 "${SOURCE_ROOT}/ptdiag/__init__.py" "${STAGING_RELEASE}/ptdiag/__init__.py"
install -m 0644 "${SOURCE_ROOT}/ptdiag/assbin.py" "${STAGING_RELEASE}/ptdiag/assbin.py"
install -m 0644 "${SOURCE_ROOT}/ptdiag/logs.py" "${STAGING_RELEASE}/ptdiag/logs.py"

0
packaging/install-dependencies.sh Executable file → Normal file
View File

View File

@@ -1095,6 +1095,7 @@ def _run_and_watch(
detected_at_ns: int | None = None
timed_out = False
stop_reason = "process-exited"
failure_reason: str | None = None
with runner_log_path.open("a", encoding="utf-8") as runner_log:
try:
process = subprocess.Popen(
@@ -1151,18 +1152,26 @@ def _run_and_watch(
current_state = state
current_analysis = analysis.to_summary()
current_text = read_current_run_log_text(state, baseline)
if analysis.load_texture_to_material_failed:
if (
analysis.load_texture_to_material_failed
or analysis.asset_pack_health_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
)
failure_reason = (
"asset-pack-health"
if analysis.asset_pack_health_failed
else "load-texture-to-material"
)
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 post_failure_deadline_ns is not None and now_ns >= post_failure_deadline_ns:
stop_reason = failure_reason or "load-texture-to-material"
_stop_process_group(process)
break
if now_ns >= deadline_ns:
timed_out = True
stop_reason = "timeout"
@@ -1216,6 +1225,9 @@ def _run_and_watch(
"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),
}
)
@@ -1244,6 +1256,7 @@ def _run_and_watch(
whispering_vale
and not timed_out
and not current_analysis.get("load_texture_to_material_failed", False)
and not current_analysis.get("asset_pack_health_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

View File

@@ -3,12 +3,13 @@
from __future__ import annotations
from dataclasses import asdict, dataclass, field
from pathlib import Path
from pathlib import Path, PurePosixPath
import re
from typing import Iterable, Mapping
DEFAULT_NATIVE_LOG_PATTERNS = ("log_*.txt", "*.log")
_ASSET_MISSING_FILE_THRESHOLD = 200
_SCENE_RE = re.compile(
r"\[S1]\s+after import\s+(?P<scene>[^:]+):\s*"
@@ -23,6 +24,28 @@ _FAILED_IMPORT_RE = re.compile(
r"Failed to import scene:\s*(?P<scene>.*)$",
re.IGNORECASE,
)
_FAILED_TEXTURE_RE = re.compile(
r"Failed to find texture\s*:\s*(?P<path>.*)$",
re.IGNORECASE,
)
_FAILED_EFFECT_RE = re.compile(
r"Failed to find effect\s*:\s*(?P<path>.*)$",
re.IGNORECASE,
)
_MISSING_IMAGE_RE = re.compile(
r"Missing file for image\s*:\s*(?P<path>.*)$",
re.IGNORECASE,
)
_FAILED_PACK_IMAGE_RE = re.compile(
r"Failed to LoadPackImage asset for\s*(?P<path>.*)$",
re.IGNORECASE,
)
_FAILED_CREATE_EFFECT_RE = re.compile(
r"Failed to create effect\s*:\s*(?P<path>.*?)"
r"(?:\s+0x[0-9a-f]+)?\s*$",
re.IGNORECASE,
)
_OUT_OF_MEMORY_RE = re.compile(r"E_OUTOFMEMORY|Ran out of memory", re.IGNORECASE)
_MODULE_CODE_SYNC_RE = re.compile(
r"Module Code Sync Error\s*\(\[(?P<left>-?\d+)]\s*!=\s*"
r"\[(?P<right>-?\d+)]\)",
@@ -67,11 +90,49 @@ class SceneImport:
return asdict(self)
@dataclass(frozen=True)
class AssetReferenceReport:
"""Manifest-aware classification of asset paths reported by the game."""
total_events: int
unique_valid_references: int
invalid_reference_events: int
catalog_available: bool
managed_present_paths: tuple[str, ...]
managed_missing_paths: tuple[str, ...]
publisher_unmanaged_paths: tuple[str, ...]
unclassified_paths: tuple[str, ...]
@property
def managed_integrity_failed(self) -> bool:
return bool(self.managed_missing_paths)
def to_dict(self) -> dict[str, object]:
return {
**asdict(self),
"managed_integrity_failed": self.managed_integrity_failed,
}
@dataclass
class NativeLogAnalysis:
scenes: dict[str, list[SceneImport]] = field(default_factory=dict)
events: list[LogEvent] = field(default_factory=list)
failed_files: list[LogEvent] = field(default_factory=list)
failed_file_find_count: int = 0
failed_file_load_count: int = 0
failed_file_read_count: int = 0
failed_asset_file_count: int = 0
failed_texture_file_count: int = 0
failed_effect_file_count: int = 0
asset_reference_events: list[LogEvent] = field(default_factory=list)
missing_texture_reference_count: int = 0
missing_effect_reference_count: int = 0
missing_image_reference_count: int = 0
missing_pack_image_reference_count: int = 0
effect_creation_failures: list[LogEvent] = field(default_factory=list)
out_of_memory_events: list[LogEvent] = field(default_factory=list)
out_of_memory_effect_paths: list[str] = field(default_factory=list)
failed_import_scenes: list[str] = field(default_factory=list)
load_texture_to_material_failed: bool = False
import_scene_threw: bool = False
@@ -83,6 +144,18 @@ class NativeLogAnalysis:
module_code_sync_delta: int | None = None
banned_code_1: bool = False
@property
def runtime_out_of_memory(self) -> bool:
return bool(self.out_of_memory_events)
@property
def asset_pack_health_failed(self) -> bool:
return (
self.failed_asset_file_count >= _ASSET_MISSING_FILE_THRESHOLD
or self.failed_texture_file_count >= _ASSET_MISSING_FILE_THRESHOLD // 3
or self.failed_effect_file_count >= max(25, _ASSET_MISSING_FILE_THRESHOLD // 8)
)
def latest_scene(self, name: str) -> SceneImport | None:
wanted = _scene_key(name)
for scene_name, imports in self.scenes.items():
@@ -98,6 +171,24 @@ class NativeLogAnalysis:
},
"load_texture_to_material_failed": self.load_texture_to_material_failed,
"import_scene_threw": self.import_scene_threw,
"failed_file_find_count": self.failed_file_find_count,
"failed_file_load_count": self.failed_file_load_count,
"failed_file_read_count": self.failed_file_read_count,
"failed_asset_file_count": self.failed_asset_file_count,
"failed_texture_file_count": self.failed_texture_file_count,
"failed_effect_file_count": self.failed_effect_file_count,
"asset_reference_event_count": len(self.asset_reference_events),
"missing_texture_reference_count": self.missing_texture_reference_count,
"missing_effect_reference_count": self.missing_effect_reference_count,
"missing_image_reference_count": self.missing_image_reference_count,
"missing_pack_image_reference_count": self.missing_pack_image_reference_count,
"effect_creation_failures": [
event.to_dict() for event in self.effect_creation_failures
],
"runtime_out_of_memory": self.runtime_out_of_memory,
"out_of_memory_count": len(self.out_of_memory_events),
"out_of_memory_effect_paths": self.out_of_memory_effect_paths,
"asset_pack_health_failed": self.asset_pack_health_failed,
"failed_import_scenes": self.failed_import_scenes,
"failed_files": [event.to_dict() for event in self.failed_files],
"stage_without_new_stage": self.stage_without_new_stage,
@@ -229,6 +320,7 @@ def parse_current_run_log(
def parse_native_log_lines(lines: Iterable[str]) -> NativeLogAnalysis:
analysis = NativeLogAnalysis()
latest_out_of_memory_line: int | None = None
for line_number, raw_line in enumerate(lines, start=1):
raw = raw_line.rstrip("\r\n")
timestamp = _timestamp(raw)
@@ -291,6 +383,65 @@ def parse_native_log_lines(lines: Iterable[str]) -> NativeLogAnalysis:
event = LogEvent(f"failed_{operation}_file", line_number, raw, timestamp, value)
analysis.failed_files.append(event)
analysis.events.append(event)
if operation == "find":
analysis.failed_file_find_count += 1
elif operation == "load":
analysis.failed_file_load_count += 1
else:
analysis.failed_file_read_count += 1
if _is_game_asset_path(value):
analysis.asset_reference_events.append(event)
analysis.failed_asset_file_count += 1
if _is_texture_path(value):
analysis.failed_texture_file_count += 1
if _is_effect_path(value):
analysis.failed_effect_file_count += 1
for kind, pattern, count_field in (
("missing_texture_reference", _FAILED_TEXTURE_RE, "missing_texture_reference_count"),
("missing_effect_reference", _FAILED_EFFECT_RE, "missing_effect_reference_count"),
("missing_image_reference", _MISSING_IMAGE_RE, "missing_image_reference_count"),
(
"missing_pack_image_reference",
_FAILED_PACK_IMAGE_RE,
"missing_pack_image_reference_count",
),
):
match = pattern.search(raw)
if match is None:
continue
value = match.group("path").strip()
event = LogEvent(kind, line_number, raw, timestamp, value)
analysis.asset_reference_events.append(event)
analysis.events.append(event)
setattr(analysis, count_field, getattr(analysis, count_field) + 1)
break
if _OUT_OF_MEMORY_RE.search(raw) is not None:
latest_out_of_memory_line = line_number
event = LogEvent("out_of_memory", line_number, raw, timestamp)
analysis.out_of_memory_events.append(event)
analysis.events.append(event)
failed_create_effect = _FAILED_CREATE_EFFECT_RE.search(raw)
if failed_create_effect is not None:
value = failed_create_effect.group("path").strip()
event = LogEvent(
"failed_create_effect",
line_number,
raw,
timestamp,
value,
)
analysis.effect_creation_failures.append(event)
analysis.events.append(event)
if (
value
and latest_out_of_memory_line is not None
and line_number - latest_out_of_memory_line <= 2
and value not in analysis.out_of_memory_effect_paths
):
analysis.out_of_memory_effect_paths.append(value)
if "Stage without new stage" in raw:
analysis.stage_without_new_stage = True
@@ -303,6 +454,80 @@ def parse_native_log_lines(lines: Iterable[str]) -> NativeLogAnalysis:
return analysis
def classify_asset_references(
analysis: NativeLogAnalysis,
managed_paths: Iterable[str] | None,
*,
source_game: Path | None = None,
) -> AssetReferenceReport:
"""Classify logged paths without treating publisher omissions as corruption."""
if managed_paths is None:
managed: dict[str, str] | None = None
else:
managed_lookup: dict[str, str] = {}
for path in managed_paths:
normalized = normalize_asset_reference(path)
if normalized is not None:
managed_lookup[normalized.casefold()] = normalized
managed = managed_lookup
unique: dict[str, str] = {}
invalid_events = 0
for event in analysis.asset_reference_events:
normalized = normalize_asset_reference(event.value or "")
if normalized is None:
invalid_events += 1
continue
unique.setdefault(normalized.casefold(), normalized)
managed_present: list[str] = []
managed_missing: list[str] = []
publisher_unmanaged: list[str] = []
unclassified: list[str] = []
for key in sorted(unique):
observed = unique[key]
if managed is None:
unclassified.append(observed)
continue
canonical = managed.get(key)
if canonical is None:
publisher_unmanaged.append(observed)
continue
if source_game is None or (source_game / PurePosixPath(canonical)).is_file():
managed_present.append(canonical)
else:
managed_missing.append(canonical)
return AssetReferenceReport(
total_events=len(analysis.asset_reference_events),
unique_valid_references=len(unique),
invalid_reference_events=invalid_events,
catalog_available=managed is not None,
managed_present_paths=tuple(managed_present),
managed_missing_paths=tuple(managed_missing),
publisher_unmanaged_paths=tuple(publisher_unmanaged),
unclassified_paths=tuple(unclassified),
)
def normalize_asset_reference(value: str) -> str | None:
"""Return a safe, slash-normalized relative asset path for comparison."""
if not value or "\x00" in value:
return None
normalized = re.sub(r"[\\/]+", "/", value.strip())
if not normalized or normalized.startswith("/"):
return None
parts = PurePosixPath(normalized).parts
if (
not parts
or any(part in ("", ".", "..") or ":" in part for part in parts)
):
return None
return "/".join(parts)
def _state(path: Path) -> LogFileState:
stat = path.stat()
return LogFileState(path.absolute(), stat.st_mtime_ns, stat.st_size, stat.st_ino)
@@ -319,3 +544,61 @@ def _scene_key(value: str) -> str:
def _timestamp(line: str) -> str | None:
match = _TIMESTAMP_RE.match(line)
return match.group("timestamp") if match is not None else None
def _is_game_asset_path(path: str) -> bool:
lower = path.casefold()
if not lower:
return False
if lower.startswith("save/") or lower.startswith("save\\"):
return False
if lower.startswith(("game\\", "game/")):
return True
if lower.startswith(("field\\", "field/")):
return True
if lower.startswith(("image\\", "image/")):
return True
if lower.startswith(("effect\\", "effect/")):
return True
if lower.startswith(("startimage\\", "startimage/")):
return True
return (
"\\effect\\" in lower
or "\\image\\" in lower
or "\\field\\" in lower
or "\\weapons\\" in lower
or "\\char\\" in lower
or "\\sinimage\\" in lower
or "\\skill\\" in lower
or lower.endswith(".bmp")
or lower.endswith(".tga")
or lower.endswith(".png")
or lower.endswith(".jpg")
or lower.endswith(".jpeg")
or lower.endswith(".gif")
or lower.endswith(".ase")
or lower.endswith(".part")
)
def _is_texture_path(path: str) -> bool:
lower = path.casefold()
return (
lower.endswith(".bmp")
or lower.endswith(".tga")
or lower.endswith(".png")
or lower.endswith(".jpg")
or lower.endswith(".jpeg")
or lower.endswith(".gif")
or lower.endswith(".dds")
)
def _is_effect_path(path: str) -> bool:
lower = path.casefold()
return (
"\\effect\\" in lower
or lower.startswith("effect\\")
or lower.startswith("effect/")
or lower.endswith(".part")
)

View File

@@ -4,4 +4,4 @@ from .paths import AppPaths
__all__ = ["AppPaths", "__version__"]
__version__ = "0.9.0"
__version__ = "0.9.5"

View File

@@ -15,6 +15,7 @@ from typing import Callable, Sequence
from ptdiag.logs import (
LogFileState,
classify_asset_references,
newest_current_native_log,
parse_current_run_log,
snapshot_native_logs,
@@ -33,6 +34,8 @@ from .depot import (
DownloadOptions,
SyncProgress,
read_cached_depot_launch_catalog,
read_current_depot_file_catalog,
preflight_managed_game,
select_depot_start_option,
sync_managed_game,
)
@@ -879,6 +882,10 @@ def _play_locked(
)
source = _source_game(paths, settings, game_dir)
overlay = None
depot_file_catalog = None
depot_preflight = None
if game_dir is None and not dry_run:
depot_file_catalog = read_current_depot_file_catalog(paths)
_print_operation_status(
as_json,
1,
@@ -900,6 +907,23 @@ def _play_locked(
"another game client is active, but the prepared runtime view is missing"
)
elif not dry_run:
if game_dir is None:
if depot_file_catalog is not None:
depot_preflight = preflight_managed_game(
paths,
catalog=depot_file_catalog,
)
if depot_preflight is not None and not depot_preflight.ready:
details = "; ".join(
f"{issue.path}: {issue.reason}"
for issue in depot_preflight.issues[:3]
)
raise LauncherError(
"the managed game integrity check found "
f"{depot_preflight.issue_count} missing or changed file(s)"
+ (f" ({details})" if details else "")
+ "; run `pristontale-eu update --repair` before playing"
)
overlay = prepare_runtime_view(paths, source)
_print_operation_status(as_json, 2, 4, "Ensuring the compatibility runtime is available…")
umu_bootstrap = (
@@ -933,9 +957,19 @@ def _play_locked(
native_log_directory,
patterns=("log_*.txt",),
)
crash_directory = paths.game_save / "crash"
crash_baseline = snapshot_native_logs(
crash_directory,
patterns=("*.dat", "*.dmp"),
)
game_started_ns = time.time_ns()
if before_game is not None:
before_game()
managed_asset_paths = (
[entry.path for entry in depot_file_catalog.files]
if depot_file_catalog is not None
else None
)
try:
outcome = run_game(
paths,
@@ -952,6 +986,8 @@ def _play_locked(
native_log_directory,
native_log_baseline,
game_started_ns,
managed_asset_paths=managed_asset_paths,
source_game=source if managed_asset_paths is not None else None,
)
if current_game_log is not None and not as_json:
_print_interrupted_game_log(current_game_log)
@@ -960,6 +996,13 @@ def _play_locked(
native_log_directory,
native_log_baseline,
game_started_ns,
managed_asset_paths=managed_asset_paths,
source_game=source if managed_asset_paths is not None else None,
)
current_game_crash = _current_game_crash(
crash_directory,
crash_baseline,
game_started_ns,
)
runtime_issues = _runtime_issue_summary(outcome.stderr)
if (
@@ -972,22 +1015,37 @@ def _play_locked(
elif runtime_issues["dbghelp_symdeferred_assertion_count"]:
runtime_issues["recommended_one_variable_comparison"] = "umu"
report = {
"ok": outcome.returncode == 0,
"ok": outcome.returncode == 0 and current_game_crash is None,
"dry_run": dry_run,
"runtime": asdict(selection),
"umu_bootstrap": asdict(umu_bootstrap) if umu_bootstrap else None,
"realm": asdict(start_option),
"runtime_view": asdict(overlay) if overlay else None,
"depot_preflight": (
{
**asdict(depot_preflight),
"ready": depot_preflight.ready,
"fully_verified": depot_preflight.fully_verified,
}
if depot_preflight
else None
),
"setup_commands": [asdict(result) for result in setup_results],
"game": asdict(outcome),
"current_game_log": current_game_log,
"current_game_crash": current_game_crash,
"runtime_issues": runtime_issues,
}
if as_json or dry_run:
_emit(report, as_json)
else:
_print_game_outcome(outcome, current_game_log, runtime_issues)
return 0 if outcome.returncode == 0 else 1
_print_game_outcome(
outcome,
current_game_log,
runtime_issues,
current_game_crash,
)
return 0 if outcome.returncode == 0 and current_game_crash is None else 1
def _ensure_portable_umu_for_play(
@@ -1028,6 +1086,9 @@ def _current_game_log(
directory: Path,
baseline: dict[str, LogFileState],
started_ns: int,
*,
managed_asset_paths: Sequence[str] | None = None,
source_game: Path | None = None,
) -> dict[str, object] | None:
state = newest_current_native_log(
directory,
@@ -1038,6 +1099,11 @@ def _current_game_log(
if state is None:
return None
analysis = parse_current_run_log(state, baseline)
asset_references = classify_asset_references(
analysis,
managed_asset_paths,
source_game=source_game,
)
return {
"path": str(state.path),
"modified_at": datetime.fromtimestamp(
@@ -1046,6 +1112,15 @@ def _current_game_log(
"size": state.size,
"load_texture_to_material_failed": analysis.load_texture_to_material_failed,
"import_scene_threw": analysis.import_scene_threw,
"asset_pack_health_failed": analysis.asset_pack_health_failed,
"managed_asset_integrity_failed": asset_references.managed_integrity_failed,
"asset_references": asset_references.to_dict(),
"runtime_out_of_memory": analysis.runtime_out_of_memory,
"out_of_memory_count": len(analysis.out_of_memory_events),
"out_of_memory_effect_paths": analysis.out_of_memory_effect_paths,
"effect_creation_failures": [
event.to_dict() for event in analysis.effect_creation_failures
],
"failed_import_scenes": analysis.failed_import_scenes,
"stage_without_new_stage": analysis.stage_without_new_stage,
"module_code_sync_error": analysis.module_code_sync_error,
@@ -1058,6 +1133,16 @@ def _current_game_log(
"native-game-log-only; server-provided login-screen text is not "
"written to this log"
),
"failed_file_find_count": analysis.failed_file_find_count,
"failed_file_load_count": analysis.failed_file_load_count,
"failed_file_read_count": analysis.failed_file_read_count,
"failed_asset_file_count": analysis.failed_asset_file_count,
"failed_texture_file_count": analysis.failed_texture_file_count,
"failed_effect_file_count": analysis.failed_effect_file_count,
"missing_texture_reference_count": analysis.missing_texture_reference_count,
"missing_effect_reference_count": analysis.missing_effect_reference_count,
"missing_image_reference_count": analysis.missing_image_reference_count,
"missing_pack_image_reference_count": analysis.missing_pack_image_reference_count,
"stage_without_new_stage_events": [
event.to_dict()
for event in analysis.events
@@ -1072,6 +1157,28 @@ def _current_game_log(
}
def _current_game_crash(
directory: Path,
baseline: dict[str, LogFileState],
started_ns: int,
) -> dict[str, object] | None:
state = newest_current_native_log(
directory,
baseline,
started_ns,
patterns=("*.dat", "*.dmp"),
)
if state is None:
return None
return {
"path": str(state.path),
"modified_at": datetime.fromtimestamp(
state.mtime_ns / 1_000_000_000
).astimezone().isoformat(),
"size": state.size,
}
def _runtime_issue_summary(stderr: str) -> dict[str, object]:
folded = stderr.casefold()
futex_message = "the futex facility returned an unexpected error code"
@@ -1106,14 +1213,24 @@ def _print_game_outcome(
outcome: CommandOutcome,
current_game_log: dict[str, object] | None,
runtime_issues: dict[str, object],
current_game_crash: dict[str, object] | None = None,
) -> None:
# Keep the human-facing stream concise. CommandRunner already relayed every
# process line live; replaying its capture here made each Proton failure appear
# twice in the dashboard.
if outcome.timed_out:
if current_game_crash is not None:
crash_path = current_game_crash.get("path")
crash_path_text = crash_path if isinstance(crash_path, str) else "unknown"
print(
f"Game client crashed and wrote a dump: {crash_path_text}",
file=sys.stderr,
)
elif outcome.timed_out:
print("Game client timed out and was stopped.", file=sys.stderr)
elif outcome.returncode == 0:
print("Game client exited normally.")
elif outcome.returncode is None:
print("Game client exited without a status.", file=sys.stderr)
else:
print(f"Game client exited with status {outcome.returncode}.", file=sys.stderr)
if current_game_log is not None:
@@ -1155,13 +1272,50 @@ def _print_game_outcome(
def _print_interrupted_game_log(report: dict[str, object]) -> None:
print(f"\nCurrent game log: {report['path']}", file=sys.stderr)
log_path = report.get("path")
log_path_text = log_path if isinstance(log_path, str) else "unknown"
print(f"\nCurrent game log: {log_path_text}", file=sys.stderr)
asset_references = report.get("asset_references")
catalog_available = (
asset_references.get("catalog_available")
if isinstance(asset_references, dict)
else False
)
if report["module_code_sync_error"]:
print(
"Detected a publisher module-integrity mismatch; the game was launched "
"with publisher-pristine code in this launcher version.",
file=sys.stderr,
)
elif report["runtime_out_of_memory"]:
effects = report.get("out_of_memory_effect_paths")
effect_text = ""
if isinstance(effects, list) and effects:
effect_text = " while creating " + ", ".join(str(path) for path in effects[:3])
print(
"Detected Direct3D resource exhaustion in the 32-bit game process"
+ effect_text
+ ". Restarting the client can recover temporarily; depot repair will not "
"fix this failure.",
file=sys.stderr,
)
elif report["managed_asset_integrity_failed"]:
print(
"The game reported an asset that belongs to the publisher manifest but is "
"missing from the managed installation.",
file=sys.stderr,
)
print(
"Run `pristontale-eu update --repair` and restart.",
file=sys.stderr,
)
elif report["asset_pack_health_failed"] and not catalog_available:
print(
"The game reported many unresolved asset paths, but no matching cached "
"publisher manifest was available to classify them. Run "
"`pristontale-eu update` to refresh the manifest cache.",
file=sys.stderr,
)
elif report["load_texture_to_material_failed"]:
print(
"Detected a texture/material failure while loading the game world.",

View File

@@ -82,6 +82,8 @@ ALWAYS_VERIFY_PATHS = frozenset(
SHA1_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$")
CONTENT_RANGE_PATTERN = re.compile(r"^bytes\s+(\d+)-(\d+)/(\d+)$", re.IGNORECASE)
DEPOT_LAUNCH_CATALOG_SCHEMA = 1
DEPOT_FILE_CATALOG_SCHEMA = 1
MAX_PREFLIGHT_ISSUES = 32
class DepotError(RuntimeError):
@@ -424,6 +426,64 @@ class DepotLaunchCatalog:
default_start_option: str
@dataclass(frozen=True)
class DepotFileCatalogEntry:
"""One manifest file plus the identity recorded after verified installation."""
path: str
size: int
sha1: str
device: int
inode: int
mtime_ns: int
ctime_ns: int
verified: bool
@dataclass(frozen=True)
class DepotFileCatalog:
"""Cached manifest used for fast, network-free launch preflight checks."""
build: int
files: tuple[DepotFileCatalogEntry, ...]
created_at: str
@dataclass(frozen=True)
class DepotIntegrityIssue:
path: str
reason: str
@dataclass(frozen=True)
class DepotPreflightResult:
"""Outcome of validating cached file identities before starting the game."""
build: int
files_total: int
verified_files: int
unverified_files: int
changed_files_hashed: int
issue_count: int
issues: tuple[DepotIntegrityIssue, ...]
duration_ms: int
@property
def ready(self) -> bool:
return self.issue_count == 0
@property
def fully_verified(self) -> bool:
return self.ready and self.unverified_files == 0
@dataclass(frozen=True)
class _ScannedGameEntry:
path: Path
state: os.stat_result
kind: str
def launch_catalog_from_snapshot(snapshot: DepotSnapshot) -> DepotLaunchCatalog:
return DepotLaunchCatalog(
build=snapshot.build,
@@ -508,6 +568,206 @@ def read_cached_depot_launch_catalog(path: Path) -> DepotLaunchCatalog | None:
)
def cache_depot_file_catalog(
path: Path,
game_directory: Path,
snapshot: DepotSnapshot,
*,
verified_paths: Iterable[str],
previous: DepotFileCatalog | None = None,
) -> DepotFileCatalog:
"""Record identities without rehashing files already verified by the updater."""
root = _prepare_game_root(game_directory, create=False)
scanned_files = _scan_game_tree(root)
explicitly_verified = {verified_path.casefold() for verified_path in verified_paths}
previous_files = (
{entry.path.casefold(): entry for entry in previous.files}
if previous is not None
else {}
)
entries: list[DepotFileCatalogEntry] = []
for file in snapshot.files:
scanned = scanned_files.get(file.path)
if scanned is None:
raise DepotError(f"cannot cache missing depot file identity: {file.path}")
state = scanned.state
if scanned.kind != "file" or state.st_size != file.size:
raise DepotError(f"cannot cache invalid depot file identity: {file.path}")
prior = previous_files.get(file.path.casefold())
verified = file.path.casefold() in explicitly_verified
if (
not verified
and prior is not None
and prior.verified
and prior.size == file.size
and prior.sha1 == file.sha1
and _catalog_identity_matches(prior, state)
):
verified = True
entries.append(_catalog_entry(file, state, verified=verified))
catalog = DepotFileCatalog(
build=snapshot.build,
files=tuple(entries),
created_at=datetime.now().astimezone().isoformat(),
)
_write_json(
path,
_depot_file_catalog_payload(catalog),
compact=True,
)
return catalog
def read_cached_depot_file_catalog(path: Path) -> DepotFileCatalog | None:
"""Read and strictly validate the local manifest/identity cache."""
try:
value = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(value, dict):
return None
if value.get("schema") != DEPOT_FILE_CATALOG_SCHEMA:
return None
build = _positive_int(value.get("build"), "cached depot file build")
created_at = _string(value.get("created_at"), "cached depot file timestamp")
raw_files = value.get("files")
if not isinstance(raw_files, list):
return None
entries: list[DepotFileCatalogEntry] = []
seen: set[str] = set()
for raw in raw_files:
entry = _parse_depot_file_catalog_entry(raw)
if entry.path.casefold() in seen:
return None
seen.add(entry.path.casefold())
entries.append(entry)
except (OSError, UnicodeError, json.JSONDecodeError, DepotError):
return None
return DepotFileCatalog(build=build, files=tuple(entries), created_at=created_at)
def read_current_depot_file_catalog(paths: AppPaths) -> DepotFileCatalog | None:
"""Return the file catalog only when it matches the last completed sync."""
catalog = read_cached_depot_file_catalog(paths.depot_file_catalog)
if catalog is None:
return None
if _completed_depot_build(paths.native_depot_state) != catalog.build:
return None
return catalog
def preflight_managed_game(
paths: AppPaths,
*,
catalog: DepotFileCatalog | None = None,
) -> DepotPreflightResult | None:
"""Check all managed paths cheaply and hash only files whose identity changed."""
catalog = catalog or read_current_depot_file_catalog(paths)
if catalog is None:
return None
if _completed_depot_build(paths.native_depot_state) != catalog.build:
return None
started = time.perf_counter()
root = _prepare_game_root(paths.source_game, create=False)
if not root.is_dir():
return DepotPreflightResult(
build=catalog.build,
files_total=len(catalog.files),
verified_files=0,
unverified_files=sum(not entry.verified for entry in catalog.files),
changed_files_hashed=0,
issue_count=1,
issues=(DepotIntegrityIssue(".", "managed game directory is missing"),),
duration_ms=_elapsed_milliseconds(started),
)
try:
scanned_files = _scan_game_tree(root)
except DepotError as exc:
return DepotPreflightResult(
build=catalog.build,
files_total=len(catalog.files),
verified_files=0,
unverified_files=sum(not entry.verified for entry in catalog.files),
changed_files_hashed=0,
issue_count=1,
issues=(DepotIntegrityIssue(".", str(exc)),),
duration_ms=_elapsed_milliseconds(started),
)
verified_files = 0
unverified_files = 0
changed_files_hashed = 0
issue_count = 0
issues: list[DepotIntegrityIssue] = []
updated_entries: list[DepotFileCatalogEntry] = []
cache_changed = False
for entry in catalog.files:
reason = None
state: os.stat_result | None = None
scanned = scanned_files.get(entry.path)
if scanned is None:
reason = "missing"
elif scanned.kind != "file":
reason = scanned.kind
elif scanned.state.st_size != entry.size:
reason = "size mismatch"
else:
state = scanned.state
updated_entry = entry
if reason is None and state is not None:
if not entry.verified:
unverified_files += 1
elif _catalog_identity_matches(entry, state):
verified_files += 1
else:
changed_files_hashed += 1
try:
assert scanned is not None
matches = _sha1(scanned.path).casefold() == entry.sha1
except DepotError as exc:
reason = f"cannot verify changed file: {exc}"
else:
if matches:
verified_files += 1
updated_entry = _catalog_entry_from_cached(entry, state)
cache_changed = True
else:
reason = "hash mismatch"
if reason is not None:
issue_count += 1
if len(issues) < MAX_PREFLIGHT_ISSUES:
issues.append(DepotIntegrityIssue(entry.path, reason))
updated_entries.append(updated_entry)
if cache_changed:
updated_catalog = DepotFileCatalog(
build=catalog.build,
files=tuple(updated_entries),
created_at=datetime.now().astimezone().isoformat(),
)
_write_json(
paths.depot_file_catalog,
_depot_file_catalog_payload(updated_catalog),
compact=True,
)
return DepotPreflightResult(
build=catalog.build,
files_total=len(catalog.files),
verified_files=verified_files,
unverified_files=unverified_files,
changed_files_hashed=changed_files_hashed,
issue_count=issue_count,
issues=tuple(issues),
duration_ms=_elapsed_milliseconds(started),
)
@dataclass(frozen=True)
class PlannedDownload:
file: DepotFile
@@ -522,6 +782,7 @@ class UpdatePlan:
deletes: tuple[str, ...]
files_total: int
bytes_total: int
hash_verified_paths: tuple[str, ...] = ()
@dataclass(frozen=True)
@@ -564,6 +825,8 @@ class DepotSyncResult:
repair: bool
dry_run: bool
completed_at: str
files_planned: int = 0
bytes_planned: int = 0
@dataclass(frozen=True)
@@ -955,6 +1218,7 @@ def plan_update(
root = _prepare_game_root(game_directory, create=False)
downloads: list[PlannedDownload] = []
hash_verified_paths: list[str] = []
for file in snapshot.files:
target = _safe_target(root, file.path)
if target.is_symlink():
@@ -976,6 +1240,8 @@ def plan_update(
):
if _sha1(target).casefold() != file.sha1.casefold():
downloads.append(PlannedDownload(file, "hash mismatch"))
else:
hash_verified_paths.append(file.path)
deletes: list[str] = []
for relative_path in snapshot.deleted_files:
@@ -992,6 +1258,7 @@ def plan_update(
deletes=tuple(deletes),
files_total=len(snapshot.files),
bytes_total=sum(item.file.size for item in downloads),
hash_verified_paths=tuple(hash_verified_paths),
)
@@ -1009,6 +1276,7 @@ def sync_managed_game(
paths.create_base_directories()
depot_client = client or DepotClient()
previous_file_catalog = read_current_depot_file_catalog(paths)
_emit_progress(progress, "manifest", 0, 0, 0, 0, None)
snapshot = depot_client.fetch_snapshot()
quarantine = QuarantineResult(0, None)
@@ -1059,6 +1327,17 @@ def sync_managed_game(
created=created,
quarantine=quarantine,
)
verified_paths = {
*plan.hash_verified_paths,
*(download.file.path for download in plan.downloads),
}
cache_depot_file_catalog(
paths.depot_file_catalog,
root,
snapshot,
verified_paths=verified_paths,
previous=previous_file_catalog,
)
_write_json(paths.native_depot_state, asdict(result))
cache_depot_launch_catalog(paths.depot_launch_catalog_cache, snapshot)
_emit_progress(
@@ -1422,6 +1701,162 @@ def _apply_create_if_missing(
return created
def _catalog_entry(
file: DepotFile,
state: os.stat_result,
*,
verified: bool,
) -> DepotFileCatalogEntry:
return DepotFileCatalogEntry(
path=file.path,
size=file.size,
sha1=file.sha1.casefold(),
device=state.st_dev,
inode=state.st_ino,
mtime_ns=state.st_mtime_ns,
ctime_ns=state.st_ctime_ns,
verified=verified,
)
def _scan_game_tree(root: Path) -> dict[str, _ScannedGameEntry]:
"""Inventory files without following symlinks or resolving every manifest path."""
entries: dict[str, _ScannedGameEntry] = {}
def scan(directory: Path, relative_parent: PurePosixPath) -> None:
children = _scan_directory(directory)
for child in children:
relative = relative_parent / child.name
key = relative.as_posix()
try:
state = child.stat(follow_symlinks=False)
except OSError as exc:
raise DepotError(f"cannot inspect managed game path {key}: {exc}") from exc
path = Path(child.path)
if stat.S_ISLNK(state.st_mode):
entries[key] = _ScannedGameEntry(path, state, "symbolic link")
elif stat.S_ISDIR(state.st_mode):
scan(path, relative)
elif stat.S_ISREG(state.st_mode):
entries[key] = _ScannedGameEntry(path, state, "file")
else:
entries[key] = _ScannedGameEntry(path, state, "not a regular file")
scan(root, PurePosixPath())
return entries
def _scan_directory(directory: Path) -> list[os.DirEntry[str]]:
try:
with os.scandir(directory) as iterator:
return list(iterator)
except OSError as exc:
raise DepotError(f"cannot scan managed game directory {directory}: {exc}") from exc
def _catalog_entry_from_cached(
entry: DepotFileCatalogEntry,
state: os.stat_result,
) -> DepotFileCatalogEntry:
return DepotFileCatalogEntry(
path=entry.path,
size=entry.size,
sha1=entry.sha1,
device=state.st_dev,
inode=state.st_ino,
mtime_ns=state.st_mtime_ns,
ctime_ns=state.st_ctime_ns,
verified=True,
)
def _catalog_identity_matches(
entry: DepotFileCatalogEntry,
state: os.stat_result,
) -> bool:
return (
entry.device == state.st_dev
and entry.inode == state.st_ino
and entry.size == state.st_size
and entry.mtime_ns == state.st_mtime_ns
and entry.ctime_ns == state.st_ctime_ns
)
def _depot_file_catalog_payload(catalog: DepotFileCatalog) -> dict[str, object]:
"""Use compact rows because this cache contains tens of thousands of files."""
return {
"schema": DEPOT_FILE_CATALOG_SCHEMA,
"build": catalog.build,
"created_at": catalog.created_at,
"files": [
[
entry.path,
entry.size,
entry.sha1,
entry.device,
entry.inode,
entry.mtime_ns,
entry.ctime_ns,
entry.verified,
]
for entry in catalog.files
],
}
def _parse_depot_file_catalog_entry(value: object) -> DepotFileCatalogEntry:
if not isinstance(value, list) or len(value) != 8:
raise DepotError("cached depot file entry must be an eight-value row")
path = _normalise_relative_path(value[0], "cached depot file path")
size = _non_negative_int(value[1], f"cached depot file size for {path}")
sha1 = _string(value[2], f"cached depot file hash for {path}").casefold()
if not SHA1_PATTERN.fullmatch(sha1):
raise DepotError(f"invalid cached SHA-1 hash for {path}")
identity: dict[str, int] = {}
for field_name, field_value in zip(
("device", "inode", "mtime_ns", "ctime_ns"),
value[3:7],
strict=True,
):
identity[field_name] = _non_negative_int(
field_value,
f"cached depot file {field_name} for {path}",
)
verified = value[7]
if not isinstance(verified, bool):
raise DepotError(f"cached depot verification flag is invalid for {path}")
return DepotFileCatalogEntry(
path=path,
size=size,
sha1=sha1,
device=identity["device"],
inode=identity["inode"],
mtime_ns=identity["mtime_ns"],
ctime_ns=identity["ctime_ns"],
verified=verified,
)
def _completed_depot_build(path: Path) -> int | None:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return None
if not isinstance(value, dict) or value.get("dry_run") is not False:
return None
build = value.get("build")
if isinstance(build, bool) or not isinstance(build, int) or build <= 0:
return None
return build
def _elapsed_milliseconds(started: float) -> int:
return max(0, round((time.perf_counter() - started) * 1000))
def _sync_result(
snapshot: DepotSnapshot,
plan: UpdatePlan,
@@ -1441,6 +1876,8 @@ def _sync_result(
build=snapshot.build,
version=snapshot.version,
files_total=plan.files_total,
files_planned=len(plan.downloads),
bytes_planned=plan.bytes_total,
files_downloaded=files_downloaded,
bytes_downloaded=bytes_downloaded,
files_deleted=files_deleted,
@@ -1919,14 +2356,16 @@ def _headers() -> dict[str, str]:
return {"User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)"}
def _write_json(path: Path, value: object) -> None:
def _write_json(path: Path, value: object, *, compact: bool = False) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
try:
temporary.write_text(
json.dumps(value, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
payload = (
json.dumps(value, separators=(",", ":"), sort_keys=True)
if compact
else json.dumps(value, indent=2, sort_keys=True)
)
temporary.write_text(payload + "\n", encoding="utf-8")
temporary.replace(path)
except BaseException:
temporary.unlink(missing_ok=True)

View File

@@ -84,7 +84,7 @@ def inspect_host(environment: Mapping[str, str] | None = None) -> DoctorReport:
"gamescope",
shutil.which("gamescope"),
False,
"optional manual game-container diagnostics",
"optional fixed-surface scaling for a stable resizable game window",
),
ToolStatus(
"cabextract",

View File

@@ -1,19 +1,114 @@
"""Native X11 command construction for the managed game window."""
"""Command construction for the managed game window."""
from __future__ import annotations
import os
from pathlib import Path
import shutil
import sys
from typing import Sequence
from typing import Mapping, Sequence
from .settings import Settings
MIN_GAME_CANVAS_DIMENSION = 320
MAX_GAME_CANVAS_DIMENSION = 16384
MAX_GAME_CONFIG_BYTES = 1024 * 1024
class GameboxConfigurationError(ValueError):
"""Raised when stable outer scaling cannot use the game configuration."""
def find_gamescope(
environment: Mapping[str, str] | None = None,
) -> str | None:
"""Find Gamescope using the environment inherited by the game launch."""
env = os.environ if environment is None else environment
return shutil.which("gamescope", path=env.get("PATH"))
def read_game_canvas_size(game_config: Path) -> tuple[int, int]:
"""Read the exact fixed render size advertised in ``game.ini``."""
try:
content = game_config.read_bytes()
except OSError as exc:
raise GameboxConfigurationError(
f"could not read game window configuration {game_config}: {exc}"
) from exc
if len(content) > MAX_GAME_CONFIG_BYTES:
raise GameboxConfigurationError(
f"game window configuration is unexpectedly large: {game_config}"
)
if b"\x00" in content:
raise GameboxConfigurationError(
f"game window configuration uses an unsupported encoding: {game_config}"
)
try:
text = content.decode("utf-8-sig")
except UnicodeDecodeError as utf8_error:
try:
text = content.decode("cp1252")
except UnicodeDecodeError:
raise GameboxConfigurationError(
f"game window configuration uses an unsupported encoding: {game_config}"
) from utf8_error
values: dict[str, int] = {}
section: str | None = None
for line_number, line in enumerate(text.splitlines(), start=1):
stripped = line.strip()
if stripped.startswith("[") and stripped.endswith("]"):
section = stripped[1:-1].strip().casefold()
continue
if section != "screen" or not stripped or stripped.startswith((";", "#")):
continue
left, separator, right = line.partition("=")
if not separator:
continue
key = left.strip().casefold()
if key not in {"width", "height"}:
continue
if key in values:
raise GameboxConfigurationError(
f"duplicate [Screen] {key.title()} in {game_config}"
)
try:
value = int(right.strip(), 10)
except ValueError as exc:
raise GameboxConfigurationError(
f"invalid [Screen] {key.title()} on line {line_number} of {game_config}"
) from exc
if value < MIN_GAME_CANVAS_DIMENSION or value > MAX_GAME_CANVAS_DIMENSION:
raise GameboxConfigurationError(
f"[Screen] {key.title()} must be between "
f"{MIN_GAME_CANVAS_DIMENSION} and {MAX_GAME_CANVAS_DIMENSION}"
)
values[key] = value
if set(values) != {"width", "height"}:
raise GameboxConfigurationError(
f"game window configuration has no complete [Screen] Width/Height pair: "
f"{game_config}"
)
return values["width"], values["height"]
def build_gamebox_command(
game_command: Sequence[str],
settings: Settings,
*,
game_size: tuple[int, int] | None = None,
gamescope_path: str | None = None,
) -> list[str]:
if not game_command:
raise ValueError("game command cannot be empty")
use_gamescope = (
settings.allow_window_resizing
and gamescope_path is not None
and game_size is not None
)
command = [
sys.executable,
"-m",
@@ -23,7 +118,48 @@ def build_gamebox_command(
]
if settings.borderless:
command.append("--borderless")
if use_gamescope:
command.append("--gamescope-scaling")
if settings.suppress_unresponsive_warnings:
command.append("--suppress-wm-ping")
command.extend(("--", *[str(value) for value in game_command]))
return command
if not use_gamescope:
return command
assert gamescope_path is not None
assert game_size is not None
width, height = game_size
if not (
MIN_GAME_CANVAS_DIMENSION <= width <= MAX_GAME_CANVAS_DIMENSION
and MIN_GAME_CANVAS_DIMENSION <= height <= MAX_GAME_CANVAS_DIMENSION
):
raise ValueError("game size is outside the supported Gamescope range")
gamescope_command = [
sys.executable,
"-m",
"ptlinux.gamescope_runtime",
"--width",
str(width),
"--height",
str(height),
"--",
gamescope_path,
"-w",
str(width),
"-h",
str(height),
"-W",
str(width),
"-H",
str(height),
# Gamescope recalculates ``fit`` from each committed game buffer, so an
# in-game resolution change keeps its own aspect ratio without cropping.
"-S",
"fit",
"-F",
"linear",
]
if settings.borderless:
gamescope_command.append("--borderless")
gamescope_command.extend(("--", *command))
return gamescope_command

View File

@@ -46,8 +46,13 @@ _X_SUBSTRUCTURE_REDIRECT_MASK = 1 << 20
_X_WM_STATE_ICONIC = 3
_NET_WM_STATE_REMOVE = 0
_NET_WM_STATE_SOURCE_APPLICATION = 1
_NET_WM_BYPASS_COMPOSITOR_PREFER_COMPOSITING = 2
WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS = 1.0
GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS = 0.25
_P_MIN_SIZE = 1 << 4
_P_MAX_SIZE = 1 << 5
_P_RESIZE_INC = 1 << 6
_P_BASE_SIZE = 1 << 8
_MOTIF_HINTS_FUNCTIONS = 1 << 0
_MOTIF_HINTS_DECORATIONS = 1 << 1
_MOTIF_ALL = 1 << 0
@@ -147,6 +152,26 @@ def _normal_hints_without_fixed_size(
return tuple(updated)
def _normal_hints_with_resize_increments(
hints: Sequence[int],
current_size: tuple[int, int],
increments: tuple[int, int],
) -> tuple[int, ...]:
"""Add resize increments while preserving every unrelated size hint."""
updated = list(hints)
if len(updated) < 18:
updated.extend(0 for _ in range(18 - len(updated)))
width, height = current_size
width_increment, height_increment = increments
updated[0] |= _P_RESIZE_INC | _P_BASE_SIZE
updated[9] = width_increment
updated[10] = height_increment
updated[15] = width % width_increment
updated[16] = height % height_increment
return tuple(updated)
class GameboxRuntimeError(RuntimeError):
"""Raised when the game command or X11 mapper cannot be started."""
@@ -257,6 +282,69 @@ class _X11:
window, title = selected
return self._client_window(window), title
def find_window_by_pid(self, expected_pid: int) -> int | None:
"""Find a top-level host window created by one native process."""
queue = self._root_windows()
visited: set[int] = set()
while queue and len(visited) < 4096:
window = queue.pop(0)
if window in visited:
continue
visited.add(window)
pid_values = self._long_property(window, b"_NET_WM_PID", 1)
if pid_values and pid_values[0] == expected_pid:
return window
queue.extend(self._child_windows(window))
return None
def set_resize_increments(
self,
window: int,
current_size: tuple[int, int],
increments: tuple[int, int],
) -> bool:
"""Quantize host resize requests without replacing other WM hints."""
if any(value <= 0 for value in (*current_size, *increments)):
raise ValueError("window sizes and resize increments must be positive")
hints = self._long_property(window, b"WM_NORMAL_HINTS", 18) or ()
updated = _normal_hints_with_resize_increments(
hints,
current_size,
increments,
)
self._replace_long_property(
window,
b"WM_NORMAL_HINTS",
b"WM_SIZE_HINTS",
updated,
)
self.library.XSync(self.display, 0)
confirmed = self._long_property(window, b"WM_NORMAL_HINTS", 18)
return bool(
confirmed
and len(confirmed) >= 17
and confirmed[0] & _P_RESIZE_INC
and confirmed[9:11] == increments
)
def prefer_composited_presentation(self, window: int) -> bool:
"""Keep the host compositor active for this window at every size."""
self._replace_long_property(
window,
b"_NET_WM_BYPASS_COMPOSITOR",
b"CARDINAL",
(_NET_WM_BYPASS_COMPOSITOR_PREFER_COMPOSITING,),
)
self.library.XSync(self.display, 0)
return self._long_property(
window,
b"_NET_WM_BYPASS_COMPOSITOR",
1,
) == (_NET_WM_BYPASS_COMPOSITOR_PREFER_COMPOSITING,)
def _root_windows(self) -> list[int]:
root = self.library.XDefaultRootWindow(self.display)
return self._child_windows(root)
@@ -466,7 +554,20 @@ class _X11:
and updated_normal_hints != normal_hints
)
motif_changed = motif_hints is not None and updated_motif_hints != motif_hints
if not normal_changed and not motif_changed:
allowed_actions = self._long_property(window, b"_NET_WM_ALLOWED_ACTIONS", 64)
allowed_actions_with_scaling = self._allowed_actions_with_scaling(
allowed_actions,
)
allowed_actions_changed = (
allowed_actions is not None
and allowed_actions_with_scaling is not None
and allowed_actions_with_scaling != allowed_actions
)
if (
not normal_changed
and not motif_changed
and not allowed_actions_changed
):
return not confirm or self.window_allows_scaling(window)
confirmed = False
@@ -485,6 +586,13 @@ class _X11:
b"_MOTIF_WM_HINTS",
updated_motif_hints,
)
if allowed_actions_changed and allowed_actions_with_scaling is not None:
self._replace_long_property(
window,
b"_NET_WM_ALLOWED_ACTIONS",
b"_ATOM",
allowed_actions_with_scaling,
)
if not confirm:
self.library.XFlush(self.display)
confirmed = True
@@ -515,6 +623,16 @@ class _X11:
)
except Exception as exc:
restore_errors.append(exc)
if allowed_actions_changed and allowed_actions is not None:
try:
self._replace_long_property(
window,
b"_NET_WM_ALLOWED_ACTIONS",
b"_ATOM",
allowed_actions,
)
except Exception as exc:
restore_errors.append(exc)
try:
self.library.XSync(self.display, 0)
except Exception as exc:
@@ -546,6 +664,36 @@ class _X11:
vertical,
}.issubset(actions)
def _allowed_actions_with_scaling(
self,
actions: tuple[int, ...] | None,
) -> tuple[int, ...] | None:
if actions is None:
return None
resize = self._intern_atom(
b"_NET_WM_ACTION_RESIZE",
only_if_exists=True,
)
horizontal = self._intern_atom(
b"_NET_WM_ACTION_MAXIMIZE_HORZ",
only_if_exists=True,
)
vertical = self._intern_atom(
b"_NET_WM_ACTION_MAXIMIZE_VERT",
only_if_exists=True,
)
if resize == 0 or horizontal == 0 or vertical == 0:
return actions
updated_actions = list(actions)
known_actions = set(actions)
for atom in (resize, horizontal, vertical):
if atom not in known_actions:
updated_actions.append(atom)
known_actions.add(atom)
return tuple(updated_actions)
def _wait_for_scaling_actions(self, window: int) -> bool:
deadline = time.monotonic() + WINDOW_ACTION_CONFIRM_TIMEOUT_SECONDS
while not self.window_allows_scaling(window):
@@ -595,6 +743,7 @@ class _X11:
watched_atoms = {
self._intern_atom(b"WM_NORMAL_HINTS", only_if_exists=False),
self._intern_atom(b"_MOTIF_WM_HINTS", only_if_exists=False),
self._intern_atom(b"_NET_WM_ALLOWED_ACTIONS", only_if_exists=False),
}
changed = False
events_read = 0
@@ -823,6 +972,11 @@ class _X11:
"""Raise the mapped game client and give it keyboard focus."""
self.library.XMapRaised(self.display, window)
self.restore_input_focus(window)
def restore_input_focus(self, window: int) -> None:
"""Give a mapped client keyboard focus without raising its host window."""
self.library.XSetInputFocus(
self.display,
window,
@@ -831,6 +985,17 @@ class _X11:
)
self.library.XSync(self.display, 0)
def window_is_focused(self, window: int) -> bool:
focused = ctypes.c_ulong()
revert_to = ctypes.c_int()
if self.library.XGetInputFocus(
self.display,
ctypes.byref(focused),
ctypes.byref(revert_to),
) != 1:
return False
return int(focused.value) == window
def activate_window(self, window: int) -> bool:
"""Ask the window manager to restore and activate one client."""
@@ -1111,6 +1276,12 @@ class _X11:
ctypes.c_ulong,
]
self.library.XSetInputFocus.restype = ctypes.c_int
self.library.XGetInputFocus.argtypes = [
ctypes.c_void_p,
ctypes.POINTER(ctypes.c_ulong),
ctypes.POINTER(ctypes.c_int),
]
self.library.XGetInputFocus.restype = ctypes.c_int
self.library.XSendEvent.argtypes = [
ctypes.c_void_p,
ctypes.c_ulong,
@@ -1250,6 +1421,48 @@ def _focus_game_window(mapper: _X11, window: int) -> None:
)
def _repair_game_window_focus(mapper: _X11, window: int) -> bool:
try:
if mapper.window_is_focused(window):
return True
except Exception:
return False
try:
mapper.activate_window(window)
time.sleep(WINDOW_ACTIVATION_DELAY_SECONDS)
mapper.focus_window(window)
return mapper.window_is_focused(window)
except Exception as exc:
print(
f"Warning: could not repair game-window focus ({exc}); continuing "
"normally",
file=sys.stderr,
flush=True,
)
return False
def _repair_gamescope_game_window_focus(mapper: _X11, window: int) -> bool:
"""Restore nested focus without asking the desktop to activate Gamescope."""
try:
if mapper.window_is_focused(window):
return True
except Exception:
return False
try:
mapper.restore_input_focus(window)
return mapper.window_is_focused(window)
except Exception as exc:
print(
f"Warning: could not restore nested game input focus ({exc}); "
"continuing normally",
file=sys.stderr,
flush=True,
)
return False
def _enable_game_window_scaling(mapper: _X11, window: int) -> bool:
"""Enable confirmed resize/maximize actions without risking game startup."""
@@ -1301,7 +1514,8 @@ def _maintain_game_window_scaling(
return int(process.wait())
repair_failures = 0
next_repair_at = 0.0
next_scaling_repair_at = 0.0
next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS
while process.poll() is None:
try:
hints_changed = mapper.wait_for_scaling_hint_change(window)
@@ -1313,13 +1527,17 @@ def _maintain_game_window_scaling(
flush=True,
)
break
if not hints_changed:
continue
now = time.monotonic()
if now < next_repair_at:
time.sleep(next_repair_at - now)
if now >= next_focus_repair_at:
_repair_game_window_focus(mapper, window)
next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS
if not hints_changed and now < next_scaling_repair_at:
continue
try:
mapper.enable_window_scaling(window, confirm=False)
mapper.enable_window_scaling(
window,
confirm=False,
)
except Exception as exc:
repair_failures += 1
print(
@@ -1337,7 +1555,53 @@ def _maintain_game_window_scaling(
break
else:
repair_failures = 0
next_repair_at = time.monotonic() + WINDOW_SCALING_REPAIR_INTERVAL_SECONDS
next_scaling_repair_at = time.monotonic() + WINDOW_SCALING_REPAIR_INTERVAL_SECONDS
return int(process.wait())
def _maintain_game_window_focus(
mapper: _X11,
process: subprocess.Popen[bytes],
window: int,
) -> int:
"""Keep the keyboard-focus repair while retaining Wine's fixed-size hints."""
print(
"Game window resizing is disabled to prevent repeated Direct3D "
"device resets.",
file=sys.stderr,
flush=True,
)
next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS
while process.poll() is None:
time.sleep(WINDOW_SCALING_EVENT_TIMEOUT_SECONDS)
now = time.monotonic()
if now < next_focus_repair_at:
continue
_repair_game_window_focus(mapper, window)
next_focus_repair_at = time.monotonic() + WINDOW_FOCUS_REPAIR_INTERVAL_SECONDS
return int(process.wait())
def _maintain_gamescope_game_window_focus(
mapper: _X11,
process: subprocess.Popen[bytes],
window: int,
) -> int:
"""Restore nested keyboard focus after host resize/maximize operations."""
next_focus_repair_at = (
time.monotonic() + GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS
)
while process.poll() is None:
time.sleep(GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS)
now = time.monotonic()
if now < next_focus_repair_at:
continue
_repair_gamescope_game_window_focus(mapper, window)
next_focus_repair_at = (
time.monotonic() + GAMESCOPE_FOCUS_REPAIR_INTERVAL_SECONDS
)
return int(process.wait())
@@ -1447,6 +1711,8 @@ def run_game_window(
window_title: str = WINDOW_TITLE,
*,
borderless: bool = False,
allow_window_resizing: bool = False,
gamescope_scaling: bool = False,
suppress_wm_ping: bool = False,
) -> int:
if not command:
@@ -1499,6 +1765,27 @@ def run_game_window(
and game_window_created_at is not None
and now - game_window_created_at >= WINDOW_READY_DELAY_SECONDS
):
if gamescope_scaling:
# Gamescope owns the nested child's fullscreen/maximized state and
# surface geometry. Trying to restore or nudge that window makes the
# two window managers fight and can produce visible frame flicker.
print(
f"Game window ready: {game_title or window_title}",
file=sys.stderr,
flush=True,
)
_focus_game_window(mapper, game_window)
print(
"Game presentation is isolated; Gamescope handles "
"aspect-preserving outer-window resizing.",
file=sys.stderr,
flush=True,
)
return _maintain_gamescope_game_window_focus(
mapper,
process,
game_window,
)
_prepare_game_window(
mapper,
game_window,
@@ -1517,12 +1804,30 @@ def run_game_window(
game_window,
borderless=borderless,
)
if startup_stable and not borderless:
if (
allow_window_resizing
and not gamescope_scaling
and startup_stable
and not borderless
):
return _maintain_game_window_scaling(
mapper,
process,
game_window,
)
if not borderless:
if allow_window_resizing and not startup_stable:
print(
"Warning: live window resizing was not enabled because the "
"startup window did not stabilize.",
file=sys.stderr,
flush=True,
)
return _maintain_game_window_focus(
mapper,
process,
game_window,
)
return int(process.wait())
time.sleep(0.1)
if process.poll() is None:
@@ -1531,7 +1836,7 @@ def run_game_window(
file=sys.stderr,
flush=True,
)
if game_window is not None:
if game_window is not None and not gamescope_scaling:
_prepare_game_window(
mapper,
game_window,
@@ -1547,6 +1852,12 @@ def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--window-title", default=WINDOW_TITLE)
parser.add_argument("--borderless", action="store_true")
parser.add_argument(
"--allow-window-resizing",
action="store_true",
help="enable unsafe native client resizing for diagnostics",
)
parser.add_argument("--gamescope-scaling", action="store_true")
parser.add_argument("--suppress-wm-ping", action="store_true")
parser.add_argument("command", nargs=argparse.REMAINDER)
arguments = parser.parse_args(argv)
@@ -1558,6 +1869,8 @@ def main(argv: Sequence[str] | None = None) -> int:
command,
arguments.window_title,
borderless=arguments.borderless,
allow_window_resizing=arguments.allow_window_resizing,
gamescope_scaling=arguments.gamescope_scaling,
suppress_wm_ping=arguments.suppress_wm_ping,
)
except (GameboxRuntimeError, OSError) as exc:

View File

@@ -0,0 +1,69 @@
"""Report the nested client status to the outer Gamescope supervisor."""
from __future__ import annotations
import argparse
import os
from pathlib import Path
import subprocess
import sys
from collections.abc import Sequence
def _process_exit_status(returncode: int) -> int:
if returncode < 0:
return 128 + min(-returncode, 127)
return min(returncode, 255)
def run_and_record(command: Sequence[str], status_file: Path) -> int:
if not command:
raise ValueError("nested game command cannot be empty")
try:
process = subprocess.Popen([str(value) for value in command])
except OSError as exc:
print(f"Could not start the nested game command: {exc}", file=sys.stderr)
returncode = 127
else:
try:
returncode = int(process.wait())
except BaseException:
if process.poll() is None:
process.terminate()
try:
process.wait(timeout=2.0)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
raise
temporary = status_file.with_name(f".{status_file.name}.{os.getpid()}.tmp")
try:
temporary.write_text(f"{returncode}\n", encoding="ascii")
os.replace(temporary, status_file)
except OSError as exc:
print(f"Could not report the nested game status: {exc}", file=sys.stderr)
try:
temporary.unlink()
except FileNotFoundError:
pass
return _process_exit_status(returncode)
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--status-file", required=True, type=Path)
parser.add_argument("command", nargs=argparse.REMAINDER)
arguments = parser.parse_args(argv)
command = arguments.command
if command and command[0] == "--":
command = command[1:]
try:
return run_and_record(command, arguments.status_file)
except ValueError as exc:
print(f"Nested game launch failed: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -0,0 +1,464 @@
"""Launch Gamescope with low-overhead host-window resize behavior."""
from __future__ import annotations
import argparse
import hashlib
import os
from pathlib import Path
import platform
import re
import subprocess
import sys
import tempfile
import time
from collections.abc import Mapping, Sequence
from .gamebox_runtime import GameboxRuntimeError, _X11
WINDOW_DISCOVERY_TIMEOUT_SECONDS = 10.0
WINDOW_DISCOVERY_POLL_SECONDS = 0.05
RESIZE_STEPS_PER_GAME_CANVAS = 20
MIN_RESIZE_INCREMENT = 8
MAX_RESIZE_INCREMENT = 96
CHILD_STATUS_POLL_SECONDS = 0.05
GAMESCOPE_VERSION_TIMEOUT_SECONDS = 1.0
WAYLAND_BACKING_FIX_VERSION = (3, 16, 28)
BUNDLED_GAMESCOPE_VERSION = (3, 16, 20)
BUNDLED_GAMESCOPE_FILENAME = "gamescope-3.16.20-ubuntu26.04-amd64"
BUNDLED_GAMESCOPE_SHA256 = (
"d4a082d8db0c467f65b429b6fb339b56b9a0d3e17e9c1957ede9ba1cb9a3ec90"
)
BUNDLED_GAMESCOPE_OS_ID = "ubuntu"
BUNDLED_GAMESCOPE_OS_VERSION = "26.04"
HASH_READ_SIZE = 256 * 1024
_GAMESCOPE_VERSION_PATTERN = re.compile(
r"\bgamescope\s+version\s+(\d+)\.(\d+)\.(\d+)",
re.IGNORECASE,
)
class GamescopeRuntimeError(RuntimeError):
"""Raised when the Gamescope process cannot be started."""
def resize_increments(width: int, height: int) -> tuple[int, int]:
"""Return coarse-enough size steps to avoid a swapchain rebuild storm."""
if width <= 0 or height <= 0:
raise ValueError("game canvas dimensions must be positive")
return (
min(
MAX_RESIZE_INCREMENT,
max(MIN_RESIZE_INCREMENT, width // RESIZE_STEPS_PER_GAME_CANVAS),
),
min(
MAX_RESIZE_INCREMENT,
max(MIN_RESIZE_INCREMENT, height // RESIZE_STEPS_PER_GAME_CANVAS),
),
)
def _supervised_command(
command: Sequence[str],
status_file: Path,
backend: str,
*,
force_composition: bool = False,
) -> list[str]:
"""Select a presentation backend and wrap the primary child."""
if not command:
raise ValueError("Gamescope command cannot be empty")
if backend not in {"sdl", "wayland"}:
raise ValueError(f"unsupported Gamescope backend: {backend}")
result = [
str(command[0]),
"--backend",
backend,
"--keep-alive",
]
if backend == "sdl" or force_composition:
result.append("--force-composition")
result.extend(map(str, command[1:]))
try:
separator = result.index("--")
except ValueError as exc:
raise ValueError("Gamescope command has no nested command separator") from exc
if separator == len(result) - 1:
raise ValueError("Gamescope command has no nested command")
result[separator + 1:separator + 1] = [
sys.executable,
"-m",
"ptlinux.gamescope_child_runtime",
"--status-file",
str(status_file),
"--",
]
return result
def _read_gamescope_version(executable: str) -> tuple[int, int, int] | None:
"""Read Gamescope's release tuple without delaying a normal launch."""
try:
completed = subprocess.run(
[executable, "--version"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="utf-8",
errors="replace",
timeout=GAMESCOPE_VERSION_TIMEOUT_SECONDS,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
return None
if completed.returncode != 0:
return None
match = _GAMESCOPE_VERSION_PATTERN.search(completed.stdout)
if match is None:
return None
major, minor, patch = match.groups()
return int(major), int(minor), int(patch)
def _bundled_gamescope_path() -> Path:
return Path(__file__).resolve().parent / "native" / BUNDLED_GAMESCOPE_FILENAME
def _bundled_gamescope_host_supported() -> bool:
"""Limit the distribution-specific executable to its build target."""
if sys.platform != "linux" or platform.machine().casefold() not in {
"amd64",
"x86_64",
}:
return False
try:
release = platform.freedesktop_os_release()
except OSError:
return False
return (
release.get("ID", "").casefold() == BUNDLED_GAMESCOPE_OS_ID
and release.get("VERSION_ID") == BUNDLED_GAMESCOPE_OS_VERSION
)
def _bundled_gamescope_is_usable() -> bool:
"""Verify the launcher-local executable and restore its safe archive mode."""
if not _bundled_gamescope_host_supported():
return False
path = _bundled_gamescope_path()
try:
if path.is_symlink() or not path.is_file():
return False
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(HASH_READ_SIZE), b""):
digest.update(chunk)
except OSError:
return False
if digest.hexdigest() != BUNDLED_GAMESCOPE_SHA256:
return False
if not os.access(path, os.X_OK):
try:
path.chmod(0o755)
except OSError:
return False
return os.access(path, os.X_OK)
def _select_wayland_gamescope(executable: str) -> tuple[str, bool, bool]:
"""Choose a direct-safe Gamescope or request the stable composed path."""
version = _read_gamescope_version(executable)
if version is not None and version >= WAYLAND_BACKING_FIX_VERSION:
return executable, False, False
if version == BUNDLED_GAMESCOPE_VERSION and _bundled_gamescope_is_usable():
return str(_bundled_gamescope_path()), False, True
return executable, True, False
def _wayland_socket_available(environment: Mapping[str, str]) -> bool:
"""Return whether the launch environment names an accessible Wayland socket."""
display = environment.get("WAYLAND_DISPLAY", "").strip()
if not display:
return False
display_path = Path(display)
if not display_path.is_absolute():
runtime_directory = environment.get("XDG_RUNTIME_DIR", "").strip()
if not runtime_directory:
return False
display_path = Path(runtime_directory) / display_path
try:
return display_path.is_socket()
except OSError:
return False
def _select_host_backend(environment: Mapping[str, str]) -> str:
"""Prefer Wayland's retained-buffer resize path when it is usable."""
if _wayland_socket_available(environment):
return "wayland"
return "sdl"
def _create_child_status_file() -> Path:
"""Create a private rendezvous file inherited by the nested child command."""
runtime_directory = os.environ.get("XDG_RUNTIME_DIR")
try:
file_descriptor, path = tempfile.mkstemp(
prefix="ptlinux-gamescope-status-",
dir=runtime_directory,
)
except OSError:
file_descriptor, path = tempfile.mkstemp(prefix="ptlinux-gamescope-status-")
os.close(file_descriptor)
return Path(path)
def _read_child_status(status_file: Path) -> int | None:
try:
value = status_file.read_text(encoding="ascii").strip()
except (FileNotFoundError, OSError, UnicodeError):
return None
if not value:
return None
try:
returncode = int(value, 10)
except ValueError:
return None
if returncode < -255 or returncode > 255:
return None
return returncode
def _process_exit_status(returncode: int) -> int:
if returncode < 0:
return 128 + min(-returncode, 127)
return min(returncode, 255)
def _wait_for_optimized_gamescope(
process: subprocess.Popen[bytes],
status_file: Path,
) -> int:
"""Return the real child status and bypass old Gamescope shutdown bugs."""
while True:
child_status = _read_child_status(status_file)
if child_status is not None:
if process.poll() is None:
process.kill()
process.wait()
return _process_exit_status(child_status)
gamescope_status = process.poll()
if gamescope_status is not None:
child_status = _read_child_status(status_file)
return _process_exit_status(
gamescope_status if child_status is None else child_status
)
time.sleep(CHILD_STATUS_POLL_SECONDS)
def _open_host_x11() -> _X11 | None:
"""Open the host X11 connection used by Gamescope's SDL window."""
if not os.environ.get("DISPLAY"):
return None
try:
return _X11()
except GameboxRuntimeError:
return None
def _tune_process_window(
mapper: _X11,
process: subprocess.Popen[bytes],
width: int,
height: int,
) -> tuple[int, int] | None:
"""Stabilize resizing once the SDL host window becomes available."""
deadline = time.monotonic() + WINDOW_DISCOVERY_TIMEOUT_SECONDS
while process.poll() is None and time.monotonic() < deadline:
window = mapper.find_window_by_pid(process.pid)
if window is not None:
increments = resize_increments(width, height)
increments_applied = mapper.set_resize_increments(
window,
(width, height),
increments,
)
compositing_preferred = mapper.prefer_composited_presentation(window)
if increments_applied and compositing_preferred:
return increments
time.sleep(WINDOW_DISCOVERY_POLL_SECONDS)
return None
def run_gamescope(command: Sequence[str], width: int, height: int) -> int:
"""Run Gamescope with stable host presentation for the current session."""
if not command:
raise GamescopeRuntimeError("Gamescope command cannot be empty")
resize_increments(width, height)
environment = os.environ.copy()
backend = _select_host_backend(environment)
selected_command = list(map(str, command))
force_composition = backend == "sdl"
using_bundled_gamescope = False
if backend == "wayland":
(
selected_command[0],
force_composition,
using_bundled_gamescope,
) = _select_wayland_gamescope(selected_command[0])
mapper = _open_host_x11() if backend == "sdl" else None
status_file: Path | None = None
try:
candidate_status_file = _create_child_status_file()
try:
launch_command = _supervised_command(
selected_command,
candidate_status_file,
backend,
force_composition=force_composition,
)
except ValueError:
candidate_status_file.unlink(missing_ok=True)
if mapper is not None:
mapper.close()
raise
status_file = candidate_status_file
except OSError as exc:
launch_command = [
selected_command[0],
"--backend",
backend,
]
if force_composition:
launch_command.append("--force-composition")
launch_command.extend(selected_command[1:])
print(
f"Warning: Gamescope child-status supervision is unavailable: {exc}; "
"continuing with stable presentation.",
file=sys.stderr,
flush=True,
)
if backend == "sdl" and environment.get("DISPLAY"):
environment["SDL_VIDEODRIVER"] = "x11"
# Keep Mutter on one composited presentation path at every window size.
# SDL otherwise requests compositor bypass, which can switch near
# screen-filling sizes and visibly flicker while Gamescope is scaling.
environment["SDL_VIDEO_X11_NET_WM_BYPASS_COMPOSITOR"] = "0"
try:
process = subprocess.Popen(launch_command, env=environment)
except OSError as exc:
if mapper is not None:
mapper.close()
if status_file is not None:
status_file.unlink(missing_ok=True)
raise GamescopeRuntimeError(f"could not start Gamescope: {exc}") from exc
if backend == "wayland" and not force_composition:
message = (
"Gamescope resize optimization active: native Wayland direct "
"presentation keeps live host-window changes off the game surface."
)
if using_bundled_gamescope:
message += (
" Launcher-local lazy-output, live-resize, letterbox backing, "
"and decoration-input fixes are active."
)
print(message, file=sys.stderr, flush=True)
elif backend == "wayland":
print(
"Warning: this Gamescope build lacks safe direct Wayland letterbox "
"backing. Forced composition prevents flicker, but live resizing may "
"pause; install Gamescope 3.16.28 or newer for direct presentation.",
file=sys.stderr,
flush=True,
)
elif mapper is not None:
try:
try:
increments = _tune_process_window(
mapper,
process,
width,
height,
)
except Exception as exc:
increments = None
print(
f"Warning: Gamescope resize tuning failed: {exc}; "
"continuing normally.",
file=sys.stderr,
flush=True,
)
finally:
mapper.close()
if increments is not None:
print(
"Gamescope resize optimization active: the host window remains "
"composited and size changes use "
f"{increments[0]}x{increments[1]} pixel steps.",
file=sys.stderr,
flush=True,
)
elif process.poll() is None:
print(
"Warning: Gamescope's host window could not be resize-tuned; "
"continuing normally.",
file=sys.stderr,
flush=True,
)
else:
print(
"Gamescope is using its SDL presentation fallback; host X11 resize "
"tuning is unavailable.",
file=sys.stderr,
flush=True,
)
try:
if status_file is not None:
return _wait_for_optimized_gamescope(process, status_file)
return _process_exit_status(int(process.wait()))
except BaseException:
if process.poll() is None:
process.terminate()
try:
process.wait(timeout=2.0)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
raise
finally:
if status_file is not None:
status_file.unlink(missing_ok=True)
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--width", required=True, type=int)
parser.add_argument("--height", required=True, type=int)
parser.add_argument("command", nargs=argparse.REMAINDER)
arguments = parser.parse_args(argv)
command = arguments.command
if command and command[0] == "--":
command = command[1:]
try:
return run_gamescope(command, arguments.width, arguments.height)
except (GamescopeRuntimeError, ValueError) as exc:
print(f"Gamescope launch failed: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -44,7 +44,8 @@ PATCHED_V21_GAME_DLL_SHA256 = (
RETIRED_POST_AUTH_GAME_DLL_SHA256 = (
"93cd27284eb294753da1111c99bfc9689f82b6d4a59302aee9fdbd8cb52452dd"
)
RUNTIME_VIEW_CONFIGURATION_VERSION = 2
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"),
@@ -68,6 +69,15 @@ class RuntimeFileIdentity:
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
@@ -82,22 +92,9 @@ class OverlayResult:
runtime_scene_identity: RuntimeFileIdentity
created_at: str
links_created: int
@dataclass(frozen=True)
class LegacyOverlayResult:
"""Version-one runtime manifest retained only for a one-time fast migration."""
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
created_at: str
links_created: int
writable_asset_count: int = 0
writable_asset_bytes: int = 0
writable_asset_source_fingerprint: str = ""
@dataclass(frozen=True)
@@ -173,14 +170,21 @@ def prepare_runtime_view(
paths: AppPaths,
source_game: Path,
) -> OverlayResult:
"""Build a symlink view and keep the launcher-managed source read-only."""
"""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)
reused = _reuse_runtime_view(paths, source, game_save, game_config)
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)
@@ -191,27 +195,11 @@ def prepare_runtime_view(
binary_action,
game_exe_source,
game_dll_source,
source_binary_hashes,
_source_binary_hashes,
) = _runtime_game_binary_profile(
paths,
source,
)
legacy_reuse = _reuse_legacy_runtime_view(
paths,
source,
inspection,
action,
game_save,
game_config,
binary_action,
game_exe_source,
game_dll_source,
source_binary_hashes,
input_identities,
)
if legacy_reuse is not None:
return legacy_reuse
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
@@ -242,16 +230,12 @@ def prepare_runtime_view(
game_source = source / "game"
game_target = build / "game"
game_target.mkdir()
links_created += _link_level(game_source, game_target, except_names={"maps"})
maps_source = game_source / "maps"
maps_target = game_target / "maps"
maps_target.mkdir()
links_created += _link_level(
maps_source,
maps_target,
except_names={"WhisperingVale.rovug"},
links_created += _build_runtime_game_tree(
game_source,
game_target,
writable_assets,
)
runtime_scene = maps_target / "WhisperingVale.rovug"
runtime_scene = build / WHISPERING_VALE_RELATIVE
if action == "normalize-known-profile":
normalization = normalize_wrapped_assbin_textures(
source_scene,
@@ -265,6 +249,10 @@ def prepare_runtime_view(
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)
@@ -291,6 +279,9 @@ def prepare_runtime_view(
"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,
)
@@ -477,6 +468,96 @@ def _link_level(source: Path, target: Path, *, except_names: set[str]) -> int:
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."""
@@ -687,6 +768,7 @@ def _runtime_input_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:
@@ -695,6 +777,17 @@ def _matching_cached_runtime_result(
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:
@@ -721,96 +814,14 @@ def _cached_runtime_compatibility(
)
def _reuse_legacy_runtime_view(
paths: AppPaths,
source: Path,
inspection: SceneInspection,
scene_action: str,
game_save: Path,
game_config: Path,
binary_action: str,
game_exe_source: Path,
game_dll_source: Path,
source_binary_hashes: Mapping[str, str],
input_identities: tuple[RuntimeFileIdentity, ...],
) -> OverlayResult | None:
"""Upgrade an unchanged v1 view after its original hash checks pass once."""
legacy = _read_legacy_overlay_result(paths.runtime_manifest)
if legacy is None:
return None
if legacy.source_game != str(source):
return None
if legacy.runtime_view != str(paths.runtime_view):
return None
if legacy.source_scene_sha256 != inspection.file_sha256:
return None
if legacy.scene_action != scene_action or legacy.binary_action != binary_action:
return None
if legacy.game_exe_sha256 != source_binary_hashes.get("Game.exe"):
return None
if legacy.game_dll_sha256 != source_binary_hashes.get("game.dll"):
return None
_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,
legacy.links_created,
):
return None
runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE
try:
if _sha256(runtime_scene) != legacy.runtime_scene_sha256:
return None
if _runtime_input_identities(paths, source) != input_identities:
raise OverlayError(
"game files changed while the legacy runtime view was being verified"
)
runtime_scene_identity = _runtime_file_identity(
"runtime-scene",
runtime_scene,
)
except OSError:
return None
result = OverlayResult(
source_game=legacy.source_game,
runtime_view=legacy.runtime_view,
source_scene_sha256=legacy.source_scene_sha256,
runtime_scene_sha256=legacy.runtime_scene_sha256,
scene_action=legacy.scene_action,
binary_action=legacy.binary_action,
game_exe_sha256=legacy.game_exe_sha256,
game_dll_sha256=legacy.game_dll_sha256,
input_identities=input_identities,
runtime_scene_identity=runtime_scene_identity,
created_at=legacy.created_at,
links_created=legacy.links_created,
)
_write_json(
paths.runtime_manifest,
{
"configuration_version": RUNTIME_VIEW_CONFIGURATION_VERSION,
**asdict(result),
},
)
return result
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)
result = _matching_cached_runtime_result(paths, source, writable_assets)
if result is None:
return None
game_exe_source = source / "Game.exe"
@@ -828,6 +839,7 @@ def _reuse_runtime_view(
game_exe_source,
game_dll_source,
result.links_created,
writable_assets,
):
return None
runtime_scene = paths.runtime_view / WHISPERING_VALE_RELATIVE
@@ -852,6 +864,7 @@ def _runtime_layout_ready(
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():
@@ -888,35 +901,14 @@ def _runtime_layout_ready(
):
return False
game_source = source / "game"
game_target = runtime / "game"
if game_target.is_symlink() or not game_target.is_dir():
return False
game_ready, game_links = _linked_level_ready(
game_source,
game_target,
except_names={"maps"},
special_names={"maps"},
game_ready, game_links = _runtime_game_tree_ready(
source / "game",
runtime / "game",
writable_assets,
)
if not game_ready:
return False
maps_source = game_source / "maps"
maps_target = game_target / "maps"
if maps_target.is_symlink() or not maps_target.is_dir():
return False
maps_ready, maps_links = _linked_level_ready(
maps_source,
maps_target,
except_names={"WhisperingVale.rovug"},
special_names={"WhisperingVale.rovug"},
)
if not maps_ready:
return False
runtime_scene = maps_target / "WhisperingVale.rovug"
if runtime_scene.is_symlink() or not runtime_scene.is_file():
return False
actual_links = top_level_links + game_links + maps_links + 2
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":
@@ -924,6 +916,66 @@ def _runtime_layout_ready(
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,
@@ -996,6 +1048,7 @@ def _read_overlay_result(path: Path) -> OverlayResult | None:
"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):
@@ -1005,6 +1058,17 @@ def _read_overlay_result(path: Path) -> OverlayResult | None:
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
@@ -1040,6 +1104,11 @@ def _read_overlay_result(path: Path) -> OverlayResult | None:
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,
)
@@ -1072,47 +1141,6 @@ def _parse_runtime_file_identity(value: object) -> RuntimeFileIdentity | None:
)
def _read_legacy_overlay_result(path: Path) -> LegacyOverlayResult | None:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
return None
if not isinstance(value, dict) or value.get("configuration_version") != 1:
return None
string_fields = (
"source_game",
"runtime_view",
"source_scene_sha256",
"runtime_scene_sha256",
"scene_action",
"binary_action",
"game_exe_sha256",
"game_dll_sha256",
"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)
or links_created < 0
):
return None
return LegacyOverlayResult(
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"],
created_at=value["created_at"],
links_created=links_created,
)
def _runtime_game_binary_profile(
paths: AppPaths,
source: Path,

View File

@@ -185,6 +185,12 @@ class AppPaths:
return self.cache / "depot-launch-catalog.json"
@property
def depot_file_catalog(self) -> Path:
"""Verified manifest paths and cheap identities used before Play."""
return self.state / "native-depot-files.json"
@property
def news_cache(self) -> Path:
return self.cache / "news.json"

View File

@@ -18,7 +18,12 @@ from typing import Mapping, Sequence
from .command import CommandOutcome, CommandRunner, OutputLineFilter
from .container_host import container_host_error, inspect_container_host
from .depot import DEFAULT_DEPOT_START_OPTION, DepotStartOption
from .gamebox import build_gamebox_command
from .gamebox import (
GameboxConfigurationError,
build_gamebox_command,
find_gamescope,
read_game_canvas_size,
)
from .paths import AppPaths
from .settings import Settings
from .umu import find_umu_launcher
@@ -376,6 +381,18 @@ def game_runtime_ready(paths: AppPaths, selection: RuntimeSelection) -> bool:
return False
def _read_game_canvas_size_or_warn(game_config: Path) -> tuple[int, int] | None:
try:
return read_game_canvas_size(game_config)
except GameboxConfigurationError as exc:
print(
f"Warning: {exc}; using a fixed game window.",
file=sys.stderr,
flush=True,
)
return None
def run_game(
paths: AppPaths,
settings: Settings,
@@ -441,10 +458,39 @@ def run_game(
environment["PROTON_VERB"] = (
"runinprefix" if concurrent_prefix_launch else "waitforexitandrun"
)
if use_gamebox:
game_command = build_gamebox_command(game_command, settings)
if extra_environment:
environment.update(extra_environment)
if use_gamebox:
gamescope_path = None
game_size = None
if settings.allow_window_resizing:
gamescope_environment = os.environ.copy()
gamescope_environment.update(environment)
gamescope_path = find_gamescope(gamescope_environment)
if gamescope_path is None:
print(
"Warning: a resizable Gamescope window was requested, but "
"Gamescope was not found; using a fixed game window.",
file=sys.stderr,
flush=True,
)
else:
game_size = _read_game_canvas_size_or_warn(paths.game_config)
if game_size is not None:
width, height = game_size
print(
"Optimized stable resize enabled through Gamescope: the game "
f"starts at {width}x{height} and keeps its aspect ratio while "
"the outer window scales.",
file=sys.stderr,
flush=True,
)
game_command = build_gamebox_command(
game_command,
settings,
game_size=game_size,
gamescope_path=gamescope_path,
)
environment["PYTHONPATH"] = _runtime_python_path(
os.environ.get("PYTHONPATH"),
)

View File

@@ -776,6 +776,7 @@ def _stage_and_install_release(
raise LauncherUpdateError(
f"launcher release destination is not a directory: {destination}"
)
_validate_release_directory(destination)
if _tree_digest(destination) != _tree_digest(staging):
raise LauncherUpdateError(
f"launcher release {manifest.version} already exists with different contents"
@@ -837,6 +838,11 @@ def _extract_release_archive(archive: Path, destination: Path, version: str) ->
raise LauncherUpdateError(
f"launcher archive member is group/world writable: {member.name!r}"
)
permissions = member.mode & 0o777
if member.isfile() and permissions not in {0o644, 0o755}:
raise LauncherUpdateError(
f"launcher archive member has unsupported permissions: {member.name!r}"
)
target = destination.joinpath(*relative.parts)
if member.isdir():
target.mkdir(parents=True, exist_ok=True, mode=0o755)
@@ -866,7 +872,7 @@ def _extract_release_archive(archive: Path, destination: Path, version: str) ->
raise LauncherUpdateError(
f"launcher payload was incomplete: {member.name!r}"
)
target.chmod(0o644)
target.chmod(permissions)
def _validate_release_directory(release: Path) -> str:
@@ -911,6 +917,16 @@ def _validate_release_directory(release: Path) -> str:
):
if not (release / required).is_file():
raise LauncherUpdateError(f"launcher release is missing {required}")
bundled_gamescope = (
release
/ "ptlinux/native/gamescope-3.16.20-ubuntu26.04-amd64"
)
if bundled_gamescope.exists():
mode = bundled_gamescope.stat().st_mode
if not stat.S_ISREG(mode) or not mode & 0o111:
raise LauncherUpdateError(
"launcher release contains a non-executable Gamescope compatibility binary"
)
return version
@@ -1014,15 +1030,20 @@ def _archive_matches(path: Path, manifest: UpdateManifest) -> bool:
def _tree_digest(root: Path) -> str:
digest = hashlib.sha256()
root_mode = root.stat().st_mode
digest.update(b"R\x00" + str(root_mode & 0o777).encode("ascii") + b"\x00")
for item in sorted(root.rglob("*"), key=lambda value: value.relative_to(root).as_posix()):
relative = item.relative_to(root).as_posix().encode("utf-8")
if item.is_symlink():
raise LauncherUpdateError(f"launcher release contains a symbolic link: {item}")
if item.is_dir():
digest.update(b"D\x00" + relative + b"\x00")
elif item.is_file():
state = item.stat()
mode = str(state.st_mode & 0o777).encode("ascii")
if stat.S_ISDIR(state.st_mode):
digest.update(b"D\x00" + relative + b"\x00" + mode + b"\x00")
elif stat.S_ISREG(state.st_mode):
digest.update(b"F\x00" + relative + b"\x00")
digest.update(str(item.stat().st_size).encode("ascii") + b"\x00")
digest.update(mode + b"\x00")
digest.update(str(state.st_size).encode("ascii") + b"\x00")
with item.open("rb") as handle:
for chunk in iter(lambda: handle.read(READ_SIZE), b""):
digest.update(chunk)

View File

@@ -45,6 +45,7 @@ class Settings:
game_source: str | None = None
download_speed_limit_mbps: float | None = None
low_bandwidth_mode: bool = False
allow_window_resizing: bool = False
@property
def source_override(self) -> Path | None:
@@ -154,6 +155,11 @@ def _settings_from_object(value: dict[str, object]) -> Settings:
realm=_string(value, "realm", "Babel").strip(),
ui_scale=_string(value, "ui_scale", "auto"),
borderless=_migrated_borderless(value),
allow_window_resizing=_boolean(
value,
"allow_window_resizing",
False,
),
suppress_unresponsive_warnings=_boolean(
value,
"suppress_unresponsive_warnings",

View File

@@ -37,6 +37,11 @@ PROTON_SYNC_LABELS: dict[str, str] = {
"Esync": "esync",
}
SETTINGS_STACKED_LAYOUT_WIDTH = 720
SETTINGS_STACKED_FOOTER_WIDTH = 640
SETTINGS_MINIMUM_WIDTH = 620
SETTINGS_MINIMUM_HEIGHT = 520
def open_settings_dialog(
root: Any,
@@ -74,7 +79,7 @@ def open_settings_dialog(
dialog = tk.Toplevel(root)
dialog.withdraw()
dialog.title("PristonTale EU — Settings")
dialog.resizable(False, False)
dialog.resizable(True, True)
dialog.configure(background="#0c0908")
dialog.transient(root)
apply_window_icon(dialog)
@@ -121,16 +126,57 @@ def open_settings_dialog(
font=("Sans", 9),
).pack(anchor="w", pady=(2, 0))
content = ttk.Frame(
settings_chrome.content,
style="Root.TFrame",
padding=(16, 14, 16, 16),
body = ttk.Frame(settings_chrome.content, style="Root.TFrame")
body.pack(fill="both", expand=True)
body.columnconfigure(0, weight=1)
body.rowconfigure(0, weight=1)
settings_viewport = ttk.Frame(body, name="settings_viewport", style="Root.TFrame")
settings_viewport.grid(
row=0,
column=0,
sticky="nsew",
padx=(16, 10),
pady=(14, 8),
)
settings_viewport.columnconfigure(0, weight=1)
settings_viewport.rowconfigure(0, weight=1)
settings_canvas = tk.Canvas(
settings_viewport,
name="settings_canvas",
background="#0c0908",
borderwidth=0,
highlightthickness=0,
takefocus=True,
yscrollincrement=max(1, round(24 * launcher_ui_scale)),
)
settings_canvas.grid(row=0, column=0, sticky="nsew")
settings_scrollbar = ttk.Scrollbar(
settings_viewport,
name="settings_scrollbar",
orient="vertical",
style="News.Vertical.TScrollbar",
command=settings_canvas.yview,
)
settings_scrollbar.grid(row=0, column=1, sticky="ns", padx=(6, 0))
settings_canvas.configure(yscrollcommand=settings_scrollbar.set)
content = ttk.Frame(
settings_canvas,
name="settings_content",
style="Root.TFrame",
padding=(0, 0, 0, 8),
)
content_window = settings_canvas.create_window(
(0, 0),
window=content,
anchor="nw",
)
content.pack(fill="both", expand=True)
content.columnconfigure(0, weight=1, uniform="settings-card")
content.columnconfigure(1, weight=1, uniform="settings-card")
borderless_value = tk.BooleanVar(value=current_settings.borderless)
allow_window_resizing_value = tk.BooleanVar(
value=current_settings.allow_window_resizing
)
ui_scale_value = tk.StringVar(
value=setting_label(current_settings.ui_scale, UI_SCALE_LABELS)
)
@@ -265,6 +311,7 @@ def open_settings_dialog(
updated_settings = update_settings(
paths,
borderless=borderless_value.get(),
allow_window_resizing=allow_window_resizing_value.get(),
ui_scale=setting_value(ui_scale_value.get(), UI_SCALE_LABELS),
suppress_unresponsive_warnings=suppress_unresponsive_value.get(),
runtime=setting_value(runtime_value.get(), RUNTIME_LABELS),
@@ -298,15 +345,20 @@ def open_settings_dialog(
display_card.grid(row=0, column=0, sticky="nsew", padx=(0, 6))
display_card.columnconfigure(0, weight=1)
display_card.columnconfigure(1, weight=1)
ttk.Label(
display_hint = ttk.Label(
display_card,
text=(
"Decorated game windows become freely resizable with persistent maximize "
"support after startup. Borderless mode removes all host decorations."
"Gamescope can resize the desktop window while keeping the game's 32-bit "
"Direct3D surface fixed. The game keeps its aspect ratio as the window "
"resizes, with black bars when the window has a different shape. Maximize "
"the window or press Super+F to toggle fullscreen. "
"If Gamescope is unavailable, the launcher keeps the native game window "
"at its configured size."
),
style="SettingsHint.TLabel",
wraplength=360,
).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11))
)
display_hint.grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11))
presentation = ttk.Frame(display_card, style="SettingsCard.TFrame")
presentation.grid(row=1, column=0, columnspan=2, sticky="ew")
ttk.Checkbutton(
@@ -317,6 +369,14 @@ def open_settings_dialog(
cursor="hand2",
takefocus=True,
).pack(fill="x")
ttk.Checkbutton(
presentation,
text="Use a resizable scaled window (Gamescope)",
variable=allow_window_resizing_value,
style="Settings.TCheckbutton",
cursor="hand2",
takefocus=True,
).pack(fill="x", pady=(7, 0))
ttk.Checkbutton(
presentation,
text="Suppress false 'not responding' warnings",
@@ -344,12 +404,13 @@ def open_settings_dialog(
rendering_card.grid(row=0, column=1, sticky="nsew", padx=(6, 0))
rendering_card.columnconfigure(0, weight=1)
rendering_card.columnconfigure(1, weight=1)
ttk.Label(
rendering_hint = ttk.Label(
rendering_card,
text="Choose which compatibility runtime launches the proven game setup.",
style="SettingsHint.TLabel",
wraplength=360,
).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11))
)
rendering_hint.grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 11))
add_choice(
rendering_card,
1,
@@ -495,29 +556,193 @@ def open_settings_dialog(
choose_game_source,
)
footer = ttk.Frame(content, style="SettingsFooter.TFrame", padding=(12, 9))
footer.grid(row=3, column=0, columnspan=2, sticky="ew", pady=(12, 0))
footer = ttk.Frame(
body,
name="settings_footer",
style="SettingsFooter.TFrame",
padding=(12, 9),
)
footer.grid(row=1, column=0, sticky="ew", padx=16, pady=(0, 16))
footer.columnconfigure(0, weight=1)
ttk.Label(
footer_note = ttk.Label(
footer,
text="Only the settings shown here affect the next launcher operation.",
style="SettingsFooter.TLabel",
).grid(row=0, column=0, sticky="w")
buttons_row = ttk.Frame(footer, style="SettingsFooter.TFrame")
)
footer_note.grid(row=0, column=0, sticky="w")
buttons_row = ttk.Frame(
footer,
name="settings_buttons",
style="SettingsFooter.TFrame",
)
buttons_row.grid(row=0, column=1, sticky="e")
ttk.Button(
buttons_row,
name="settings_cancel",
text="CANCEL",
style="Secondary.TButton",
command=close_settings,
).pack(side="left")
ttk.Button(
buttons_row,
name="settings_save",
text="SAVE SETTINGS",
style="Play.TButton",
command=save_current_settings,
).pack(side="left", padx=(8, 0))
settings_layout_mode: tuple[bool, bool] | None = None
settings_layout_width: int | None = None
pending_settings_width = 1
settings_layout_after: str | None = None
def arrange_settings(viewport_width: int) -> None:
"""Reflow cards and footer controls for the available logical width."""
nonlocal settings_layout_mode, settings_layout_width
if settings_layout_width == viewport_width:
return
settings_layout_width = viewport_width
logical_width = viewport_width / max(launcher_ui_scale, 0.01)
stacked = logical_width < SETTINGS_STACKED_LAYOUT_WIDTH
footer_stacked = logical_width < SETTINGS_STACKED_FOOTER_WIDTH
card_width = (
viewport_width
if stacked
else (viewport_width - round(12 * launcher_ui_scale)) // 2
)
card_wraplength = max(140, card_width - round(42 * launcher_ui_scale))
display_hint.configure(wraplength=card_wraplength)
rendering_hint.configure(wraplength=card_wraplength)
footer_note.configure(
wraplength=(
max(140, viewport_width - round(42 * launcher_ui_scale))
if footer_stacked
else max(140, viewport_width - round(300 * launcher_ui_scale))
)
)
mode = (stacked, footer_stacked)
if settings_layout_mode == mode:
return
settings_layout_mode = mode
if stacked:
display_card.grid_configure(
row=0,
column=0,
columnspan=2,
sticky="ew",
padx=0,
pady=0,
)
rendering_card.grid_configure(
row=1,
column=0,
columnspan=2,
sticky="ew",
padx=0,
pady=(12, 0),
)
downloads_card.grid_configure(row=2, pady=(12, 0))
overrides_card.grid_configure(row=3, pady=(12, 0))
else:
display_card.grid_configure(
row=0,
column=0,
columnspan=1,
sticky="nsew",
padx=(0, 6),
pady=0,
)
rendering_card.grid_configure(
row=0,
column=1,
columnspan=1,
sticky="nsew",
padx=(6, 0),
pady=0,
)
downloads_card.grid_configure(row=1, pady=(12, 0))
overrides_card.grid_configure(row=2, pady=(12, 0))
if footer_stacked:
footer_note.grid_configure(row=0, column=0, columnspan=2)
buttons_row.grid_configure(
row=1,
column=0,
columnspan=2,
pady=(8, 0),
)
else:
footer_note.grid_configure(row=0, column=0, columnspan=1)
buttons_row.grid_configure(
row=0,
column=1,
columnspan=1,
pady=0,
)
def refresh_scroll_region(_event: object | None = None) -> None:
bounds = settings_canvas.bbox("all")
if bounds is not None:
settings_canvas.configure(scrollregion=bounds)
def apply_pending_settings_layout() -> None:
nonlocal settings_layout_after
settings_layout_after = None
arrange_settings(pending_settings_width)
def resize_settings_content(event: Any) -> None:
nonlocal pending_settings_width, settings_layout_after
pending_settings_width = max(1, int(event.width))
settings_canvas.itemconfigure(content_window, width=pending_settings_width)
if settings_layout_after is None:
settings_layout_after = dialog.after_idle(apply_pending_settings_layout)
def scroll_settings(event: Any) -> str | None:
canvas_left = settings_canvas.winfo_rootx()
canvas_top = settings_canvas.winfo_rooty()
pointer_x = dialog.winfo_pointerx()
pointer_y = dialog.winfo_pointery()
if not (
canvas_left <= pointer_x < canvas_left + settings_canvas.winfo_width()
and canvas_top <= pointer_y < canvas_top + settings_canvas.winfo_height()
):
return None
if getattr(event.widget, "winfo_class", lambda: "")() == "TCombobox":
return None
first, last = settings_canvas.yview()
if last - first >= 0.999:
return None
if getattr(event, "num", None) == 4:
units = -3
elif getattr(event, "num", None) == 5:
units = 3
else:
delta = int(getattr(event, "delta", 0))
if delta == 0:
return None
steps = max(1, min(4, abs(delta) // 120 or 1))
units = -steps if delta > 0 else steps
settings_canvas.yview_scroll(units, "units")
return "break"
content.bind("<Configure>", refresh_scroll_region)
settings_canvas.bind("<Configure>", resize_settings_content)
dialog.bind("<MouseWheel>", scroll_settings, add="+")
dialog.bind("<Button-4>", scroll_settings, add="+")
dialog.bind("<Button-5>", scroll_settings, add="+")
settings_canvas.bind("<Home>", lambda _event: settings_canvas.yview_moveto(0.0))
settings_canvas.bind("<End>", lambda _event: settings_canvas.yview_moveto(1.0))
settings_canvas.bind(
"<Prior>",
lambda _event: settings_canvas.yview_scroll(-1, "pages"),
)
settings_canvas.bind(
"<Next>",
lambda _event: settings_canvas.yview_scroll(1, "pages"),
)
dialog.protocol("WM_DELETE_WINDOW", close_settings)
dialog.bind("<Escape>", lambda _event: close_settings())
dialog.bind("<Control-s>", lambda _event: save_current_settings())
@@ -529,6 +754,14 @@ def open_settings_dialog(
screen_width=dialog.winfo_screenwidth(),
screen_height=dialog.winfo_screenheight(),
)
minimum_width, minimum_height = fitted_window_size(
SETTINGS_MINIMUM_WIDTH,
SETTINGS_MINIMUM_HEIGHT,
scale=launcher_ui_scale,
screen_width=dialog.winfo_screenwidth(),
screen_height=dialog.winfo_screenheight(),
)
dialog.minsize(minimum_width, minimum_height)
dialog.geometry(
centered_window_geometry(
settings_width,

View File

@@ -1,6 +1,6 @@
[project]
name = "PTEU-Linux-Launcher"
version = "0.9.0"
version = "0.9.5"
description = "PristonTale EU Linux installer, clean launcher, and diagnostic harness"
requires-python = ">=3.11"
dependencies = []

View File

@@ -1,11 +1,11 @@
{
"archive_sha256": "5a11b5ef8ab8edc07b04414c1f5631fe8366b8384f70d962336433fc19d80851",
"archive_size": 296981,
"archive_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/download/v0.9.0/pristontale-eu-linux-update-v0.9.0.tar.gz",
"archive_sha256": "890443aab4cd0e33bb2e2a44ea86b3fd929488eba045b803c2c0662ce68c08ff",
"archive_size": 1410277,
"archive_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/download/v0.9.5/pristontale-eu-linux-update-v0.9.5.tar.gz",
"channel": "stable",
"minimum_updater_version": "0.5.0",
"notes_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/tag/v0.9.0",
"published_at": "2026-09-03T21:15:17Z",
"notes_url": "https://git.m4ko.net/m4ko/pristontale-eu-linux/releases/tag/v0.9.5",
"published_at": "2026-09-06T20:06:31Z",
"schema": 1,
"version": "0.9.0"
"version": "0.9.5"
}

Binary file not shown.