diff --git a/install.sh b/install.sh old mode 100755 new mode 100644 diff --git a/packaging/install-dependencies.sh b/packaging/install-dependencies.sh old mode 100755 new mode 100644 diff --git a/ptdiag/logs.py b/ptdiag/logs.py index bbf3fba..8584979 100644 --- a/ptdiag/logs.py +++ b/ptdiag/logs.py @@ -3,7 +3,7 @@ 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 @@ -24,6 +24,28 @@ _FAILED_IMPORT_RE = re.compile( r"Failed to import scene:\s*(?P.*)$", re.IGNORECASE, ) +_FAILED_TEXTURE_RE = re.compile( + r"Failed to find texture\s*:\s*(?P.*)$", + re.IGNORECASE, +) +_FAILED_EFFECT_RE = re.compile( + r"Failed to find effect\s*:\s*(?P.*)$", + re.IGNORECASE, +) +_MISSING_IMAGE_RE = re.compile( + r"Missing file for image\s*:\s*(?P.*)$", + re.IGNORECASE, +) +_FAILED_PACK_IMAGE_RE = re.compile( + r"Failed to LoadPackImage asset for\s*(?P.*)$", + re.IGNORECASE, +) +_FAILED_CREATE_EFFECT_RE = re.compile( + r"Failed to create effect\s*:\s*(?P.*?)" + 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-?\d+)]\s*!=\s*" r"\[(?P-?\d+)]\)", @@ -68,6 +90,30 @@ 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) @@ -79,6 +125,14 @@ class NativeLogAnalysis: 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 @@ -90,6 +144,10 @@ 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 ( @@ -119,6 +177,17 @@ class NativeLogAnalysis: "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], @@ -251,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) @@ -320,12 +390,59 @@ def parse_native_log_lines(lines: Iterable[str]) -> NativeLogAnalysis: 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 analysis.events.append( @@ -337,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) diff --git a/ptlinux/cli.py b/ptlinux/cli.py index 1eb76c3..a45c17f 100644 --- a/ptlinux/cli.py +++ b/ptlinux/cli.py @@ -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( @@ -1047,6 +1113,14 @@ def _current_game_log( "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, @@ -1065,6 +1139,10 @@ def _current_game_log( "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 @@ -1079,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" @@ -1113,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: @@ -1162,22 +1272,48 @@ 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["asset_pack_health_failed"]: + 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 a large number of missing game-asset files. This indicates a " - "corrupted or incomplete installation that needs repair.", + "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. If the issue " - "reappears, re-download the game files with a clean install path.", + "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"]: diff --git a/ptlinux/depot.py b/ptlinux/depot.py index 2197639..5ae6320 100644 --- a/ptlinux/depot.py +++ b/ptlinux/depot.py @@ -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) diff --git a/ptlinux/overlay.py b/ptlinux/overlay.py index 8abd502..7e5bafb 100644 --- a/ptlinux/overlay.py +++ b/ptlinux/overlay.py @@ -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, diff --git a/ptlinux/paths.py b/ptlinux/paths.py index 95edd99..a2f485a 100644 --- a/ptlinux/paths.py +++ b/ptlinux/paths.py @@ -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"