summaryrefslogtreecommitdiff
path: root/py_modules/lsfg_vk
diff options
context:
space:
mode:
Diffstat (limited to 'py_modules/lsfg_vk')
-rw-r--r--py_modules/lsfg_vk/config_schema.py2
-rw-r--r--py_modules/lsfg_vk/configuration.py76
-rw-r--r--py_modules/lsfg_vk/constants.py3
-rw-r--r--py_modules/lsfg_vk/flatpak_profile_service.py381
-rw-r--r--py_modules/lsfg_vk/flatpak_service.py589
-rw-r--r--py_modules/lsfg_vk/plugin.py112
-rw-r--r--py_modules/lsfg_vk/steam_service.py59
-rw-r--r--py_modules/lsfg_vk/wrapper_service.py225
8 files changed, 950 insertions, 497 deletions
diff --git a/py_modules/lsfg_vk/config_schema.py b/py_modules/lsfg_vk/config_schema.py
index 3816d88..bf3e174 100644
--- a/py_modules/lsfg_vk/config_schema.py
+++ b/py_modules/lsfg_vk/config_schema.py
@@ -113,6 +113,6 @@ class ConfigurationManager:
**profile,
**global_config,
})
- if config["active_in"]:
+ if name or config["active_in"]:
profiles[name] = config
return {"profiles": profiles, "global_config": global_config}
diff --git a/py_modules/lsfg_vk/configuration.py b/py_modules/lsfg_vk/configuration.py
index bec3828..17dcaf3 100644
--- a/py_modules/lsfg_vk/configuration.py
+++ b/py_modules/lsfg_vk/configuration.py
@@ -7,7 +7,7 @@ from .runtime_service import RuntimeService
class ConfigurationService(BaseService):
- """Controller-facing adapter over upstream lsfg-vk profiles."""
+ FLATPAK_PROFILE_PREFIX = "flatpak:"
def __init__(self, logger=None, runtime_service: RuntimeService = None):
super().__init__(logger)
@@ -47,6 +47,13 @@ class ConfigurationService(BaseService):
(None, None),
)
+ @classmethod
+ def flatpak_profile_name(cls, app_id: str) -> str:
+ value = str(app_id).strip()
+ if not value:
+ raise ValueError("Flatpak application ID is required")
+ return f"{cls.FLATPAK_PROFILE_PREFIX}{value}"
+
@staticmethod
def _public_config(config: Dict[str, Any]) -> Dict[str, Any]:
return ConfigurationManager.validate_config(config)
@@ -87,6 +94,64 @@ class ConfigurationService(BaseService):
except Exception as error:
return self._error_response(dict, str(error), appid=str(appid), config=None)
+ def get_flatpak_config(self, app_id: str) -> Dict[str, Any]:
+ try:
+ data = self._get_profile_data()
+ name = self.flatpak_profile_name(app_id)
+ raw = data["profiles"].get(name)
+ return self._success_response(
+ dict,
+ app_id=str(app_id),
+ profile=name,
+ exists=raw is not None,
+ config=self._public_config(raw) if raw is not None else None,
+ global_config=dict(data["global_config"]),
+ )
+ except Exception as error:
+ return self._error_response(dict, str(error), app_id=str(app_id), config=None, exists=False)
+
+ def update_flatpak_config(self, app_id: str, config: Dict[str, Any]) -> Dict[str, Any]:
+ try:
+ data = self._get_profile_data()
+ name = self.flatpak_profile_name(app_id)
+ merged_config = {**data["global_config"], **config}
+ if not config.get("dll"):
+ merged_config["dll"] = data["global_config"].get("dll", "")
+ validated = self._public_config(merged_config)
+ validated["active_in"] = []
+ data["global_config"] = {
+ "dll": validated["dll"],
+ "no_fp16": validated["no_fp16"],
+ }
+ data["profiles"][name] = validated
+ self._save_profile_data(data)
+ return self._success_response(dict, app_id=str(app_id), profile=name, exists=True, config=validated)
+ except Exception as error:
+ return self._error_response(dict, str(error), app_id=str(app_id), config=None, exists=False)
+
+ def reset_flatpak_config(self, app_id: str) -> Dict[str, Any]:
+ try:
+ data = self._get_profile_data()
+ name = self.flatpak_profile_name(app_id)
+ data["profiles"].pop(name, None)
+ self._save_profile_data(data)
+ return self._success_response(dict, app_id=str(app_id), profile=name, exists=False)
+ except Exception as error:
+ return self._error_response(dict, str(error), app_id=str(app_id), config=None, exists=False)
+
+ def reset_all_flatpak_configs(self) -> Dict[str, Any]:
+ try:
+ data = self._get_profile_data()
+ data["profiles"] = {
+ name: profile
+ for name, profile in data["profiles"].items()
+ if not name.startswith(self.FLATPAK_PROFILE_PREFIX)
+ }
+ self._save_profile_data(data)
+ return self._success_response(dict, global_config=dict(data["global_config"]))
+ except Exception as error:
+ return self._error_response(dict, str(error))
+
def reset_game_config(self, appid: str) -> Dict[str, Any]:
try:
data = self._get_profile_data()
@@ -101,7 +166,14 @@ class ConfigurationService(BaseService):
def reset_all_game_configs(self) -> Dict[str, Any]:
try:
data = self._get_profile_data()
- data["profiles"] = {}
+ data["profiles"] = {
+ name: profile
+ for name, profile in data["profiles"].items()
+ if not (
+ len(profile.get("active_in", [])) == 1
+ and re.fullmatch(r"-?[0-9]+", str(profile.get("active_in", [""])[0]))
+ )
+ }
self._save_profile_data(data)
return self._success_response(dict, global_config=dict(data["global_config"]), games=[])
except Exception as error:
diff --git a/py_modules/lsfg_vk/constants.py b/py_modules/lsfg_vk/constants.py
index 960d230..45ac4c5 100644
--- a/py_modules/lsfg_vk/constants.py
+++ b/py_modules/lsfg_vk/constants.py
@@ -16,9 +16,6 @@ CLI_FILENAME = "lsfg-vk-cli"
UI_FILENAME = "lsfg-vk-ui"
UI_DESKTOP_FILENAME = "gay.pancake.lsfg-vk-ui.desktop"
UI_ICON_FILENAME = "gay.pancake.lsfg-vk-ui.png"
-FLATPAK_23_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_23.08.flatpak"
-FLATPAK_24_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_24.08.flatpak"
-FLATPAK_25_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_25.08.flatpak"
STEAM_LOSSLESS_SCALING_APP_ID = "993090"
STEAM_LOSSLESS_SCALING_BRANCH = "lsfg-vk"
diff --git a/py_modules/lsfg_vk/flatpak_profile_service.py b/py_modules/lsfg_vk/flatpak_profile_service.py
new file mode 100644
index 0000000..ddec116
--- /dev/null
+++ b/py_modules/lsfg_vk/flatpak_profile_service.py
@@ -0,0 +1,381 @@
+from __future__ import annotations
+
+import re
+from typing import Any, Dict
+
+from .configuration import ConfigurationService
+from .flatpak_service import FlatpakService
+
+
+class FlatpakProfileService:
+ STATE_FIELDS = (
+ "dxvkFrameRate",
+ "disableGamescopeWsi",
+ "disableHdr",
+ "disableSteamdeckMode",
+ "disableVkbasalt",
+ "enableZink",
+ )
+ BOOLEAN_FIELDS = STATE_FIELDS[1:]
+ DXVK_FRAME_RATE_SEGMENT = re.compile(
+ r"^(?:dxvk\.maxFrameRate|dxgi\.maxFrameRate|d3d9\.maxFrameRate)\s*=",
+ re.IGNORECASE,
+ )
+
+ def __init__(
+ self,
+ flatpak_service: FlatpakService,
+ configuration_service: ConfigurationService,
+ ):
+ self.flatpak_service = flatpak_service
+ self.configuration_service = configuration_service
+
+ @classmethod
+ def default_state(cls) -> Dict[str, Any]:
+ return {
+ "dxvkFrameRate": 0,
+ "disableGamescopeWsi": True,
+ "disableHdr": True,
+ "disableSteamdeckMode": False,
+ "disableVkbasalt": False,
+ "enableZink": False,
+ }
+
+ @classmethod
+ def _validate_state(cls, raw: Any) -> Dict[str, Any]:
+ if not isinstance(raw, dict):
+ raise ValueError("Workaround state must be an object")
+ missing = [field for field in cls.STATE_FIELDS if field not in raw]
+ if missing:
+ raise ValueError("Workaround state is missing: " + ", ".join(missing))
+ state = {field: raw[field] for field in cls.STATE_FIELDS}
+ frame_rate = state["dxvkFrameRate"]
+ if isinstance(frame_rate, bool) or not isinstance(frame_rate, int) or not 0 <= frame_rate <= 60:
+ raise ValueError("Base FPS Cap must be an integer from 0 to 60")
+ for field in cls.BOOLEAN_FIELDS:
+ if type(state[field]) is not bool:
+ raise ValueError(f"{field} must be a boolean")
+ return state
+
+ def _state_entry(self, app_id: str) -> tuple[Dict[str, object], Dict[str, Any]]:
+ state = self.flatpak_service._read_state()
+ entry = state["prepared_apps"].get(app_id)
+ if not isinstance(entry, dict):
+ raise RuntimeError("Flatpak application is not owned by this plugin")
+ return state, entry
+
+ def _baseline_content(self, app_id: str, entry: Dict[str, Any]) -> str:
+ if not entry.get("override_existed"):
+ return ""
+ path = self.flatpak_service._backup_path(app_id)
+ if path.is_symlink() or not path.is_file():
+ raise RuntimeError("Flatpak override backup is unavailable")
+ return path.read_text(encoding="utf-8")
+
+ @staticmethod
+ def _environment_value(content: str, key: str) -> str:
+ section = None
+ for raw_line in content.splitlines():
+ line = raw_line.strip()
+ if line.startswith("[") and line.endswith("]"):
+ section = line[1:-1]
+ continue
+ if section != "Environment":
+ continue
+ name, separator, value = line.partition("=")
+ if separator and name == key:
+ return value
+ return ""
+
+ @classmethod
+ def _dxvk_config(cls, baseline: str, frame_rate: int) -> str:
+ existing = cls._environment_value(baseline, "DXVK_CONFIG")
+ parts = [part.strip() for part in existing.split(";") if part.strip()]
+ parts = [part for part in parts if not cls.DXVK_FRAME_RATE_SEGMENT.match(part)]
+ if frame_rate > 0:
+ parts.append(f"dxvk.maxFrameRate = {frame_rate}")
+ return "; ".join(parts)
+
+ def _restore_baseline(self, app_id: str, entry: Dict[str, Any]) -> None:
+ existed, current = self.flatpak_service._snapshot_override(app_id)
+ current_hash = self.flatpak_service._sha256(current) if existed else self.flatpak_service._sha256(b"")
+ if current_hash != entry.get("managed_sha256"):
+ raise RuntimeError("Flatpak override changed after preparation; refusing to overwrite unrelated settings")
+ path = self.flatpak_service._override_path(app_id)
+ if entry.get("override_existed"):
+ self.flatpak_service._write_file(path, self._baseline_content(app_id, entry))
+ else:
+ path.unlink(missing_ok=True)
+
+ def _apply_state(self, app_id: str, workaround_state: Dict[str, Any]) -> Dict[str, Any]:
+ workaround_state = self._validate_state(workaround_state)
+ _, entry = self._state_entry(app_id)
+ baseline = self._baseline_content(app_id, entry)
+ self._restore_baseline(app_id, entry)
+ prepared = self.flatpak_service.prepare_app(app_id)
+ if not prepared.get("success") or not prepared.get("owned"):
+ raise RuntimeError(prepared.get("error") or "Could not restore plugin-owned Flatpak preparation")
+ profile = self.configuration_service.flatpak_profile_name(app_id)
+ args = [
+ "override",
+ "--user",
+ f"--env=LSFGVK_PROFILE={profile}",
+ "--unset-env=DISABLE_LSFGVK",
+ "--unset-env=DISABLE_LSFG",
+ ]
+ if workaround_state["disableGamescopeWsi"]:
+ args.extend(["--env=ENABLE_GAMESCOPE_WSI=0", "--unset-env=DISABLE_GAMESCOPE_WSI"])
+ if workaround_state["disableHdr"]:
+ args.append("--env=DXVK_HDR=0")
+ if workaround_state["disableSteamdeckMode"]:
+ args.append("--env=SteamDeck=0")
+ if workaround_state["disableVkbasalt"]:
+ args.extend(["--env=DISABLE_VKBASALT=1", "--unset-env=ENABLE_VKBASALT"])
+ if workaround_state["enableZink"]:
+ args.extend([
+ "--env=__GLX_VENDOR_LIBRARY_NAME=mesa",
+ "--env=MESA_LOADER_DRIVER_OVERRIDE=zink",
+ "--env=GALLIUM_DRIVER=zink",
+ ])
+ dxvk_config = self._dxvk_config(baseline, workaround_state["dxvkFrameRate"])
+ if dxvk_config:
+ args.append(f"--env=DXVK_CONFIG={dxvk_config}")
+ args.append(app_id)
+ result = self.flatpak_service._run_flatpak_command(args, capture_output=True, text=True)
+ if result.returncode != 0:
+ raise OSError(result.stderr.strip() or f"Could not apply Flatpak workarounds for {app_id}")
+ existed, managed = self.flatpak_service._snapshot_override(app_id)
+ if not existed:
+ raise RuntimeError(f"Flatpak override for {app_id} was not created")
+ state = self.flatpak_service._read_state()
+ entry = state["prepared_apps"].get(app_id)
+ if not isinstance(entry, dict):
+ raise RuntimeError("Flatpak application ownership state disappeared")
+ entry["managed_sha256"] = self.flatpak_service._sha256(managed)
+ entry["workaround_state"] = workaround_state
+ self.flatpak_service._write_state(state)
+ return workaround_state
+
+ def enable_app(self, app_id: str) -> Dict[str, Any]:
+ created_profile = False
+ newly_owned = False
+ try:
+ existing = self.configuration_service.get_flatpak_config(app_id)
+ before = self.flatpak_service._read_state()
+ was_owned = app_id in before["prepared_apps"]
+ prepared = self.flatpak_service.prepare_app(app_id)
+ if not prepared.get("success"):
+ raise RuntimeError(prepared.get("error") or "Could not prepare Flatpak application")
+ if not prepared.get("owned"):
+ raise RuntimeError("Flatpak application is prepared outside this plugin and cannot be managed safely")
+ newly_owned = not was_owned
+ if not existing.get("exists"):
+ config = {
+ **self.configuration_service._public_config({}),
+ **(existing.get("global_config") or {}),
+ }
+ config["active_in"] = []
+ saved = self.configuration_service.update_flatpak_config(app_id, config)
+ if not saved.get("success"):
+ raise RuntimeError(saved.get("error") or "Could not create Flatpak profile")
+ created_profile = True
+ _, entry = self._state_entry(app_id)
+ workaround_state = self._validate_state(entry.get("workaround_state", self.default_state()))
+ self._apply_state(app_id, workaround_state)
+ return self.get_app(app_id)
+ except Exception as error:
+ if created_profile:
+ self.configuration_service.reset_flatpak_config(app_id)
+ if newly_owned:
+ self.flatpak_service.remove_app_override(app_id)
+ return {
+ "success": False,
+ "message": "",
+ "error": str(error),
+ "app_id": str(app_id),
+ "enabled": False,
+ }
+
+ def update_config(self, app_id: str, config: Dict[str, Any]) -> Dict[str, Any]:
+ try:
+ self._state_entry(app_id)
+ result = self.configuration_service.update_flatpak_config(app_id, config)
+ if not result.get("success"):
+ raise RuntimeError(result.get("error") or "Could not update Flatpak profile")
+ return self.get_app(app_id)
+ except Exception as error:
+ return {
+ "success": False,
+ "message": "",
+ "error": str(error),
+ "app_id": str(app_id),
+ "enabled": False,
+ }
+
+ def get_workaround_state(self, app_id: str) -> Dict[str, Any]:
+ try:
+ _, entry = self._state_entry(app_id)
+ state = self._validate_state(entry.get("workaround_state", self.default_state()))
+ return {
+ "success": True,
+ "message": "",
+ "error": None,
+ "app_id": app_id,
+ "state": state,
+ }
+ except Exception as error:
+ return {
+ "success": False,
+ "message": "",
+ "error": str(error),
+ "app_id": str(app_id),
+ "state": None,
+ }
+
+ def set_workaround_state(self, app_id: str, workaround_state: Dict[str, Any]) -> Dict[str, Any]:
+ try:
+ state = self._apply_state(app_id, workaround_state)
+ return {
+ "success": True,
+ "message": "",
+ "error": None,
+ "app_id": app_id,
+ "state": state,
+ }
+ except Exception as error:
+ return {
+ "success": False,
+ "message": "",
+ "error": str(error),
+ "app_id": str(app_id),
+ "state": None,
+ }
+
+ def remove_app(self, app_id: str) -> Dict[str, Any]:
+ try:
+ removed = self.flatpak_service.remove_app_override(app_id)
+ if not removed.get("success"):
+ raise RuntimeError(removed.get("error") or "Could not remove Flatpak preparation")
+ reset = self.configuration_service.reset_flatpak_config(app_id)
+ if not reset.get("success"):
+ raise RuntimeError(reset.get("error") or "Could not remove Flatpak profile")
+ return {
+ "success": True,
+ "message": "Flatpak profile removed",
+ "error": None,
+ "app_id": app_id,
+ "enabled": False,
+ }
+ except Exception as error:
+ return {
+ "success": False,
+ "message": "",
+ "error": str(error),
+ "app_id": str(app_id),
+ "enabled": True,
+ }
+
+ def get_app(self, app_id: str) -> Dict[str, Any]:
+ apps = self.get_apps()
+ if not apps.get("success"):
+ return {
+ "success": False,
+ "message": "",
+ "error": apps.get("error"),
+ "app_id": str(app_id),
+ "enabled": False,
+ }
+ app = next((item for item in apps.get("apps", []) if item.get("app_id") == app_id), None)
+ if app is None:
+ return {
+ "success": False,
+ "message": "",
+ "error": "Flatpak application is not installed",
+ "app_id": str(app_id),
+ "enabled": False,
+ }
+ return {"success": True, "message": "", "error": None, **app}
+
+ def get_apps(self) -> Dict[str, Any]:
+ try:
+ result = self.flatpak_service.get_flatpak_apps()
+ if not result.get("success"):
+ raise RuntimeError(result.get("error") or "Could not list Flatpak applications")
+ ownership = self.flatpak_service._read_state()
+ apps = []
+ for item in result.get("apps", []):
+ app_id = item["app_id"]
+ config_result = self.configuration_service.get_flatpak_config(app_id)
+ entry = ownership["prepared_apps"].get(app_id)
+ workarounds = self.default_state()
+ if isinstance(entry, dict):
+ workarounds = self._validate_state(entry.get("workaround_state", workarounds))
+ profile = self.configuration_service.flatpak_profile_name(app_id)
+ selector_ready = False
+ if item.get("prepared"):
+ shown = self.flatpak_service._run_flatpak_command(
+ ["override", "--user", "--show", app_id],
+ capture_output=True,
+ text=True,
+ )
+ selector_ready = shown.returncode == 0 and f"LSFGVK_PROFILE={profile}" in shown.stdout.splitlines()
+ apps.append({
+ **item,
+ "profile": profile,
+ "enabled": bool(item.get("owned") and config_result.get("exists") and selector_ready),
+ "config": config_result.get("config"),
+ "workarounds": workarounds,
+ })
+ return {
+ "success": True,
+ "message": result.get("message", ""),
+ "error": None,
+ "apps": apps,
+ }
+ except Exception as error:
+ return {"success": False, "message": "", "error": str(error), "apps": []}
+
+ def get_running_apps(self) -> Dict[str, Any]:
+ try:
+ state = self.flatpak_service._read_state()
+ enabled = set()
+ for app_id, entry in state["prepared_apps"].items():
+ if not isinstance(entry, dict):
+ continue
+ config = self.configuration_service.get_flatpak_config(app_id)
+ if not config.get("exists"):
+ continue
+ existed, content = self.flatpak_service._snapshot_override(app_id)
+ if not existed or self.flatpak_service._sha256(content) != entry.get("managed_sha256"):
+ continue
+ profile = self.configuration_service.flatpak_profile_name(app_id)
+ try:
+ text = content.decode("utf-8")
+ except UnicodeDecodeError:
+ continue
+ if self._environment_value(text, "LSFGVK_PROFILE") == profile:
+ enabled.add(app_id)
+ if not enabled:
+ return {"success": True, "message": "", "error": None, "apps": []}
+ result = self.flatpak_service._run_flatpak_command(
+ ["ps", "--columns=application,active,pid"],
+ capture_output=True,
+ text=True,
+ )
+ if result.returncode != 0:
+ raise OSError(result.stderr.strip() or "Could not inspect running Flatpak applications")
+ running = []
+ for line in result.stdout.splitlines():
+ fields = line.split("\t") if "\t" in line else line.split()
+ if not fields or fields[0] not in enabled:
+ continue
+ active = len(fields) > 1 and fields[1].strip().lower() in {"1", "true", "yes", "active"}
+ running.append({
+ "app_id": fields[0],
+ "active": active,
+ "pid": fields[2].strip() if len(fields) > 2 else "",
+ })
+ running.sort(key=lambda item: (not item["active"], item["app_id"]))
+ return {"success": True, "message": "", "error": None, "apps": running}
+ except Exception as error:
+ return {"success": False, "message": "", "error": str(error), "apps": []}
diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py
index f486b74..4f04382 100644
--- a/py_modules/lsfg_vk/flatpak_service.py
+++ b/py_modules/lsfg_vk/flatpak_service.py
@@ -1,7 +1,6 @@
-"""Flatpak runtime support for classified Steam targets."""
-
from __future__ import annotations
+import hashlib
import json
import os
import pwd
@@ -13,21 +12,16 @@ from pathlib import Path
from typing import Dict, Optional, Set
from .base_service import BaseService
-from .constants import (
- BIN_DIR,
- FLATPAK_23_08_FILENAME,
- FLATPAK_24_08_FILENAME,
- FLATPAK_25_08_FILENAME,
-)
class FlatpakService(BaseService):
EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk"
- SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08")
+ FLATHUB_REMOTE = "flathub"
+ SUPPORTED_RUNTIMES = ("24.08", "25.08")
DERIVED_RUNTIME_IDS = {"org.gnome.Platform", "org.kde.Platform"}
RUNTIME_METADATA_SECTION = "Extension org.freedesktop.Platform.GL"
- OWNERSHIP_FILENAME = "flatpak_extensions.json"
- OWNERSHIP_VERSION = 1
+ OWNERSHIP_FILENAME = "flatpak_state.json"
+ OWNERSHIP_VERSION = 2
APP_ID_PATTERN = re.compile(
r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$"
)
@@ -35,12 +29,17 @@ class FlatpakService(BaseService):
def __init__(self, logger=None):
super().__init__(logger)
self.flatpak_command: Optional[str] = None
+ self._verified_branches: Set[str] = set()
self._lock = threading.RLock()
@property
def ownership_path(self) -> Path:
return self.config_dir / self.OWNERSHIP_FILENAME
+ @property
+ def backup_dir(self) -> Path:
+ return self.config_dir / "flatpak-overrides"
+
def _clean_env(self) -> Dict[str, str]:
env = os.environ.copy()
env.pop("LD_LIBRARY_PATH", None)
@@ -89,13 +88,6 @@ class FlatpakService(BaseService):
return branch
@classmethod
- def runtime_branch_from_ref(cls, runtime_ref: str) -> str:
- parts = runtime_ref.strip().split("/") if isinstance(runtime_ref, str) else []
- if len(parts) != 3 or parts[0] != "org.freedesktop.Platform":
- raise ValueError(f"Unsupported Flatpak runtime reference: {runtime_ref}")
- return cls._validate_runtime(parts[2])
-
- @classmethod
def runtime_branch_from_metadata(cls, metadata: str) -> str:
section = None
versions = []
@@ -119,18 +111,8 @@ class FlatpakService(BaseService):
def _extension_ref(cls, branch: str) -> str:
return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(branch)}"
- def _bundled_extension_path(self, branch: str) -> Path:
- filename = {
- "23.08": FLATPAK_23_08_FILENAME,
- "24.08": FLATPAK_24_08_FILENAME,
- "25.08": FLATPAK_25_08_FILENAME,
- }[self._validate_runtime(branch)]
- return Path(__file__).resolve().parent.parent.parent / BIN_DIR / filename
-
def _installed_extension_branches(self, scope: Optional[str] = None) -> Set[str]:
scopes = ("user", "system") if scope is None else (scope,)
- if any(item not in ("user", "system") for item in scopes):
- raise ValueError("Flatpak installation scope must be user or system")
installed = set()
for item in scopes:
result = self._run_flatpak_command(
@@ -145,64 +127,86 @@ class FlatpakService(BaseService):
installed.add(fields[2])
return installed
- def _owned_branches(self) -> Set[str]:
- path = self.ownership_path
- if not path.exists() and not path.is_symlink():
- return set()
- if path.is_symlink() or not path.is_file():
+ def _user_extension_origin(self, branch: str) -> str:
+ result = self._run_flatpak_command(
+ ["info", "--user", "--show-origin", self._extension_ref(branch)],
+ capture_output=True,
+ text=True,
+ )
+ return result.stdout.strip() if result.returncode == 0 else ""
+
+ def _empty_state(self) -> Dict[str, object]:
+ return {
+ "version": self.OWNERSHIP_VERSION,
+ "plugin_owned_branches": [],
+ "prepared_apps": {},
+ }
+
+ def _read_state(self) -> Dict[str, object]:
+ if not self.ownership_path.exists():
+ return self._empty_state()
+ if self.ownership_path.is_symlink() or not self.ownership_path.is_file():
raise RuntimeError("Flatpak ownership metadata is not a regular file")
try:
- data = json.loads(path.read_text(encoding="utf-8"))
- branches = data.get("plugin_owned_branches")
- if data.get("version") != self.OWNERSHIP_VERSION or not isinstance(branches, list):
- raise ValueError("invalid ownership metadata")
- owned = {self._validate_runtime(branch) for branch in branches}
- if len(owned) != len(branches):
- raise ValueError("invalid ownership metadata")
- return owned
- except (OSError, json.JSONDecodeError, TypeError, ValueError) as error:
- raise RuntimeError(f"Could not trust Flatpak ownership metadata: {error}") from error
-
- def _write_owned_branches(self, branches: Set[str]) -> None:
- if not branches:
+ data = json.loads(self.ownership_path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError) as error:
+ raise RuntimeError(f"Could not read Flatpak ownership metadata: {error}") from error
+ if data.get("version") != self.OWNERSHIP_VERSION:
+ raise RuntimeError("Unsupported Flatpak ownership metadata version")
+ branches = data.get("plugin_owned_branches")
+ apps = data.get("prepared_apps")
+ if not isinstance(branches, list) or not isinstance(apps, dict):
+ raise RuntimeError("Invalid Flatpak ownership metadata")
+ for branch in branches:
+ self._validate_runtime(branch)
+ for app_id, entry in apps.items():
+ self._validate_app_id(app_id)
+ if not isinstance(entry, dict):
+ raise RuntimeError("Invalid Flatpak app ownership metadata")
+ if type(entry.get("override_existed")) is not bool:
+ raise RuntimeError("Invalid Flatpak app ownership metadata")
+ if not isinstance(entry.get("managed_sha256"), str):
+ raise RuntimeError("Invalid Flatpak app ownership metadata")
+ return data
+
+ def _write_state(self, state: Dict[str, object]) -> None:
+ branches = state.get("plugin_owned_branches", [])
+ apps = state.get("prepared_apps", {})
+ if not branches and not apps:
self.ownership_path.unlink(missing_ok=True)
+ if self.backup_dir.exists() and not any(self.backup_dir.iterdir()):
+ self.backup_dir.rmdir()
return
self._write_file(
self.ownership_path,
- json.dumps(
- {
- "version": self.OWNERSHIP_VERSION,
- "plugin_owned_branches": sorted(branches),
- },
- indent=2,
- ) + "\n",
+ json.dumps(state, indent=2, sort_keys=True) + "\n",
)
- def get_extension_status(self):
- try:
- available = self.check_flatpak_available()
- installed = self._installed_extension_branches() if available else set()
- return self._success_response(
- dict,
- "Flatpak runtime extension status retrieved" if available else "Flatpak is not available",
- available=available,
- extension_id=self.EXTENSION_ID,
- supported_branches=list(self.SUPPORTED_RUNTIMES),
- installed_branches=sorted(installed),
- )
- except Exception as error:
- return self._error_response(
- dict,
- str(error),
- available=False,
- extension_id=self.EXTENSION_ID,
- supported_branches=list(self.SUPPORTED_RUNTIMES),
- installed_branches=[],
- )
+ def _owned_branches(self, state: Optional[Dict[str, object]] = None) -> Set[str]:
+ current = state if state is not None else self._read_state()
+ return {self._validate_runtime(branch) for branch in current["plugin_owned_branches"]}
- get_flatpak_support_status = get_extension_status
+ def _override_path(self, app_id: str) -> Path:
+ return self.user_home / ".local/share/flatpak/overrides" / self._validate_app_id(app_id)
- def _resolve_runtime(self, app_id: str):
+ def _backup_path(self, app_id: str) -> Path:
+ return self.backup_dir / f"{self._validate_app_id(app_id)}.ini"
+
+ @staticmethod
+ def _sha256(content: bytes) -> str:
+ return hashlib.sha256(content).hexdigest()
+
+ def _snapshot_override(self, app_id: str) -> tuple[bool, bytes]:
+ path = self._override_path(app_id)
+ if path.is_symlink():
+ raise RuntimeError("Flatpak override path is a symlink")
+ if not path.exists():
+ return False, b""
+ if not path.is_file():
+ raise RuntimeError("Flatpak override path is not a regular file")
+ return True, path.read_bytes()
+
+ def _resolve_runtime(self, app_id: str) -> tuple[str, str]:
self._validate_app_id(app_id)
if not self.check_flatpak_available():
raise FileNotFoundError("Flatpak is not available on this system")
@@ -218,77 +222,146 @@ class FlatpakService(BaseService):
if len(parts) != 3:
raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}")
if parts[0] == "org.freedesktop.Platform":
- branch = self._validate_runtime(parts[2])
- elif parts[0] in self.DERIVED_RUNTIME_IDS:
- metadata_result = self._run_flatpak_command(
- ["info", "--show-metadata", runtime],
- capture_output=True,
- text=True,
- )
- if metadata_result.returncode != 0:
- raise OSError(
- metadata_result.stderr.strip()
- or f"Could not inspect Flatpak runtime {runtime}"
- )
- branch = self.runtime_branch_from_metadata(metadata_result.stdout)
- else:
+ return runtime, self._validate_runtime(parts[2])
+ if parts[0] not in self.DERIVED_RUNTIME_IDS:
raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}")
- return runtime, branch
+ metadata_result = self._run_flatpak_command(
+ ["info", "--show-metadata", runtime],
+ capture_output=True,
+ text=True,
+ )
+ if metadata_result.returncode != 0:
+ raise OSError(metadata_result.stderr.strip() or f"Could not inspect Flatpak runtime {runtime}")
+ return runtime, self.runtime_branch_from_metadata(metadata_result.stdout)
+
+ def _dll_directory(self) -> Path:
+ if self.config_file_path.exists():
+ try:
+ content = self.config_file_path.read_text(encoding="utf-8")
+ match = re.search(
+ r'(?m)^[ \t]*dll[ \t]*=[ \t]*"((?:\\.|[^"\\])*)"',
+ content,
+ )
+ if match:
+ configured_dll = json.loads('"' + match.group(1) + '"')
+ if configured_dll:
+ return Path(configured_dll).parent
+ except Exception:
+ pass
+ return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling"
- def resolve_app_support(self, app_id: str):
+ def _filesystem_present(self, entries: str, host_path: Path) -> bool:
+ accepted = {str(host_path)}
try:
- app_id = self._validate_app_id(app_id)
- runtime, branch = self._resolve_runtime(app_id)
- installed = self._installed_extension_branches()
- ready = branch in installed
+ accepted.add(f"~/{host_path.relative_to(self.user_home).as_posix()}")
+ except ValueError:
+ pass
+ enabled = False
+ for raw in entries.split(";"):
+ value = raw.strip()
+ if not value:
+ continue
+ denied = value.startswith("!")
+ path = value[1:] if denied else value
+ path = path.split(":", 1)[0]
+ if path in accepted:
+ if denied:
+ return False
+ enabled = True
+ return enabled
+
+ def _app_override_status(self, app_id: str) -> Dict[str, object]:
+ result = self._run_flatpak_command(
+ ["override", "--user", "--show", app_id],
+ capture_output=True,
+ text=True,
+ )
+ output = result.stdout if result.returncode == 0 else ""
+ section = None
+ filesystems = ""
+ unset_environment = set()
+ environment = {}
+ for raw_line in output.splitlines():
+ line = raw_line.strip()
+ if line.startswith("[") and line.endswith("]"):
+ section = line[1:-1]
+ continue
+ key, separator, value = line.partition("=")
+ if not separator:
+ continue
+ if section == "Context" and key == "filesystems":
+ filesystems = value
+ elif section == "Context" and key == "unset-environment":
+ unset_environment.update(item for item in value.split(";") if item)
+ elif section == "Environment":
+ environment[key] = value
+ config_ready = self._filesystem_present(filesystems, self.config_dir)
+ dll_ready = self._filesystem_present(filesystems, self._dll_directory())
+ env_ready = (
+ environment.get("LSFGVK_CONFIG") == str(self.config_file_path)
+ and environment.get("LSFGVK_FLATPAK") == "1"
+ and "DISABLE_LSFGVK" in unset_environment
+ and "DISABLE_LSFG" in unset_environment
+ )
+ return {
+ "filesystem_ready": config_ready and dll_ready,
+ "environment_ready": env_ready,
+ "prepared": config_ready and dll_ready and env_ready,
+ }
+
+ def get_extension_status(self):
+ try:
+ available = self.check_flatpak_available()
+ installed = self._installed_extension_branches() if available else set()
return self._success_response(
dict,
- f"lsfg-vk support is ready for {app_id}" if ready
- else f"lsfg-vk runtime extension {branch} is required for {app_id}",
- flatpak_app_id=app_id,
- runtime=runtime,
- runtime_branch=branch,
- support_status="ready" if ready else "needs-runtime",
- extension_installed=ready,
+ "Flatpak runtime extension status retrieved" if available else "Flatpak is not available",
+ available=available,
+ extension_id=self.EXTENSION_ID,
+ supported_branches=list(self.SUPPORTED_RUNTIMES),
installed_branches=sorted(installed),
)
- except ValueError as error:
- return self._success_response(
- dict,
- str(error),
- flatpak_app_id=app_id,
- runtime=None,
- runtime_branch=None,
- support_status="unsupported",
- extension_installed=False,
- installed_branches=[],
- error=str(error),
- )
except Exception as error:
return self._error_response(
dict,
str(error),
- flatpak_app_id=app_id,
- runtime=None,
- runtime_branch=None,
- support_status="error",
- extension_installed=False,
+ available=False,
+ extension_id=self.EXTENSION_ID,
+ supported_branches=list(self.SUPPORTED_RUNTIMES),
installed_branches=[],
)
+ get_flatpak_support_status = get_extension_status
+
def install_extension(self, branch: str):
try:
branch = self._validate_runtime(branch)
if not self.check_flatpak_available():
raise FileNotFoundError("Flatpak is not available on this system")
with self._lock:
- if branch in self._installed_extension_branches():
- return self._extension_result(branch, True, False, "already installed")
- bundle = self._bundled_extension_path(branch)
- if not bundle.is_file():
- raise FileNotFoundError(f"Bundled Flatpak extension not found at {bundle}; reinstall the plugin")
+ if branch in self._verified_branches:
+ return self._extension_result(branch, True, False, "is ready")
+ user_installed = self._installed_extension_branches("user")
+ system_installed = self._installed_extension_branches("system")
+ if branch in system_installed and branch not in user_installed:
+ return self._extension_result(branch, True, False, "is ready")
+ if branch in user_installed and self._user_extension_origin(branch) != self.FLATHUB_REMOTE:
+ result = self._run_flatpak_command(
+ ["uninstall", "--user", "--noninteractive", self._extension_ref(branch)],
+ capture_output=True,
+ text=True,
+ )
+ if result.returncode != 0:
+ raise OSError(result.stderr.strip() or "Could not replace the existing Flatpak extension")
result = self._run_flatpak_command(
- ["install", "--user", "--noninteractive", "--or-update", str(bundle)],
+ [
+ "install",
+ "--user",
+ "--noninteractive",
+ "--or-update",
+ self.FLATHUB_REMOTE,
+ f"{self.EXTENSION_ID}//{branch}",
+ ],
capture_output=True,
text=True,
)
@@ -296,9 +369,12 @@ class FlatpakService(BaseService):
raise OSError(result.stderr.strip() or "Flatpak installation failed")
if branch not in self._installed_extension_branches("user"):
raise RuntimeError(f"Flatpak install completed but {self._extension_ref(branch)} was not visible afterwards")
- owned = self._owned_branches()
+ state = self._read_state()
+ owned = self._owned_branches(state)
owned.add(branch)
- self._write_owned_branches(owned)
+ state["plugin_owned_branches"] = sorted(owned)
+ self._write_state(state)
+ self._verified_branches.add(branch)
return self._extension_result(branch, True, False, "installed")
except Exception as error:
return self._error_response(dict, str(error), runtime_branch=branch, installed=False, enabled=False)
@@ -313,8 +389,6 @@ class FlatpakService(BaseService):
)
if result.returncode != 0:
raise OSError(result.stderr.strip() or "Flatpak uninstall failed")
- if branch in self._installed_extension_branches("user"):
- raise RuntimeError(f"Flatpak uninstall completed but {self._extension_ref(branch)} is still installed")
return True
def _extension_result(self, branch: str, installed: bool, removed: bool, verb: str):
@@ -333,99 +407,226 @@ class FlatpakService(BaseService):
if not self.check_flatpak_available():
raise FileNotFoundError("Flatpak is not available on this system")
with self._lock:
- owned = self._owned_branches()
+ state = self._read_state()
+ owned = self._owned_branches(state)
if branch not in owned:
installed = branch in self._installed_extension_branches()
return self._extension_result(branch, installed, False, "preserved (not plugin-owned)")
removed = self._remove_extension(branch)
owned.remove(branch)
- self._write_owned_branches(owned)
+ state["plugin_owned_branches"] = sorted(owned)
+ self._write_state(state)
installed = branch in self._installed_extension_branches()
return self._extension_result(branch, installed, removed, "uninstalled")
except Exception as error:
- return self._error_response(
- dict,
- str(error),
- runtime_branch=branch,
- removed=False,
- installed=False,
- enabled=False,
- )
+ return self._error_response(dict, str(error), runtime_branch=branch, removed=False, installed=False, enabled=False)
def ensure_extension(self, branch: str):
- try:
- branch = self._validate_runtime(branch)
- if branch in self._installed_extension_branches():
- return self._extension_result(branch, True, False, "is ready")
- except Exception as error:
- return self._error_response(dict, str(error), runtime_branch=branch, support_status="error")
return self.install_extension(branch)
- def ensure_app_support(self, app_id: str):
- resolved = self.resolve_app_support(app_id)
- if not resolved.get("success") or resolved.get("support_status") != "needs-runtime":
- return resolved
- result = self.ensure_extension(resolved["runtime_branch"])
- if not result.get("success"):
- return self._error_response(
- dict,
- result.get("error") or "Could not install the required Flatpak runtime extension",
- flatpak_app_id=app_id,
- runtime=resolved.get("runtime"),
- runtime_branch=resolved.get("runtime_branch"),
- support_status="error",
- extension_installed=False,
- )
- return self.resolve_app_support(app_id)
-
def set_extension_enabled(self, branch: str, enabled: bool):
if type(enabled) is not bool:
return self._error_response(dict, "enabled must be a boolean", runtime_branch=branch, installed=False, enabled=False)
return self.install_extension(branch) if enabled else self.uninstall_extension(branch)
- def remove_plugin_owned_extensions(self):
+ def get_flatpak_apps(self):
+ try:
+ if not self.check_flatpak_available():
+ raise FileNotFoundError("Flatpak is not available on this system")
+ installed_extensions = self._installed_extension_branches()
+ state = self._read_state()
+ owned_apps = state["prepared_apps"]
+ result = self._run_flatpak_command(
+ ["list", "--app", "--columns=name,application"],
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+ apps = []
+ for line in result.stdout.splitlines():
+ fields = line.split("\t")
+ if len(fields) < 2:
+ continue
+ name, app_id = fields[0].strip(), fields[1].strip()
+ if not app_id:
+ continue
+ item = {
+ "app_id": app_id,
+ "app_name": name or app_id,
+ "runtime": None,
+ "runtime_branch": None,
+ "runtime_ready": False,
+ "prepared": False,
+ "owned": app_id in owned_apps,
+ "error": None,
+ }
+ try:
+ runtime, branch = self._resolve_runtime(app_id)
+ status = self._app_override_status(app_id)
+ item.update({
+ "runtime": runtime,
+ "runtime_branch": branch,
+ "runtime_ready": branch in installed_extensions,
+ "prepared": status["prepared"],
+ })
+ except Exception as error:
+ item["error"] = str(error)
+ apps.append(item)
+ apps.sort(key=lambda item: str(item["app_name"]).lower())
+ return self._success_response(dict, f"Found {len(apps)} Flatpak applications", apps=apps)
+ except Exception as error:
+ return self._error_response(dict, str(error), apps=[])
+
+ def prepare_app(self, app_id: str):
try:
+ app_id = self._validate_app_id(app_id)
+ with self._lock:
+ runtime, branch = self._resolve_runtime(app_id)
+ extension = self.ensure_extension(branch)
+ if not extension.get("success") or not extension.get("installed"):
+ raise RuntimeError(extension.get("error") or f"Could not install Flatpak runtime {branch}")
+ state = self._read_state()
+ apps = state["prepared_apps"]
+ status = self._app_override_status(app_id)
+ if status["prepared"] and app_id not in apps:
+ return self._success_response(
+ dict,
+ "Flatpak application is already prepared outside this plugin",
+ app_id=app_id,
+ runtime=runtime,
+ runtime_branch=branch,
+ prepared=True,
+ owned=False,
+ )
+ if app_id not in apps:
+ existed, original = self._snapshot_override(app_id)
+ backup = self._backup_path(app_id)
+ if existed:
+ self._write_file(backup, original.decode("utf-8"))
+ else:
+ backup.unlink(missing_ok=True)
+ apps[app_id] = {
+ "override_existed": existed,
+ "managed_sha256": "",
+ }
+ result = self._run_flatpak_command(
+ [
+ "override",
+ "--user",
+ f"--filesystem={self.config_dir}:ro",
+ f"--filesystem={self._dll_directory()}:ro",
+ f"--env=LSFGVK_CONFIG={self.config_file_path}",
+ "--env=LSFGVK_FLATPAK=1",
+ "--unset-env=DISABLE_LSFGVK",
+ "--unset-env=DISABLE_LSFG",
+ app_id,
+ ],
+ capture_output=True,
+ text=True,
+ )
+ if result.returncode != 0:
+ raise OSError(result.stderr.strip() or f"Could not prepare Flatpak app {app_id}")
+ status = self._app_override_status(app_id)
+ if not status["prepared"]:
+ raise RuntimeError(f"Flatpak preparation did not become visible for {app_id}")
+ existed, managed = self._snapshot_override(app_id)
+ if not existed:
+ raise RuntimeError(f"Flatpak override for {app_id} was not created")
+ apps[app_id]["managed_sha256"] = self._sha256(managed)
+ self._write_state(state)
+ return self._success_response(
+ dict,
+ "Flatpak application prepared for lsfg-vk",
+ app_id=app_id,
+ runtime=runtime,
+ runtime_branch=branch,
+ prepared=True,
+ owned=True,
+ )
+ except Exception as error:
+ return self._error_response(dict, str(error), app_id=app_id, prepared=False, owned=False)
+
+ def remove_app_override(self, app_id: str):
+ try:
+ app_id = self._validate_app_id(app_id)
with self._lock:
- owned = self._owned_branches()
- if not owned:
+ state = self._read_state()
+ apps = state["prepared_apps"]
+ entry = apps.get(app_id)
+ if entry is None:
return self._success_response(
dict,
- "No plugin-owned Flatpak extensions to remove",
+ "Flatpak application is not plugin-owned; existing overrides were preserved",
+ app_id=app_id,
+ prepared=self._app_override_status(app_id)["prepared"],
+ owned=False,
+ )
+ existed, current = self._snapshot_override(app_id)
+ current_hash = self._sha256(current) if existed else self._sha256(b"")
+ if current_hash != entry["managed_sha256"]:
+ raise RuntimeError(
+ "Flatpak override changed after preparation; refusing to overwrite unrelated settings"
+ )
+ override_path = self._override_path(app_id)
+ backup_path = self._backup_path(app_id)
+ if entry["override_existed"]:
+ if not backup_path.is_file() or backup_path.is_symlink():
+ raise RuntimeError("Flatpak override backup is unavailable")
+ self._write_file(override_path, backup_path.read_text(encoding="utf-8"))
+ else:
+ override_path.unlink(missing_ok=True)
+ backup_path.unlink(missing_ok=True)
+ apps.pop(app_id, None)
+ self._write_state(state)
+ return self._success_response(
+ dict,
+ "Plugin-owned Flatpak preparation removed",
+ app_id=app_id,
+ prepared=False,
+ owned=False,
+ )
+ except Exception as error:
+ return self._error_response(dict, str(error), app_id=app_id, prepared=False, owned=True)
+
+ def remove_plugin_owned_environment(self):
+ try:
+ with self._lock:
+ state = self._read_state()
+ failures = []
+ removed_apps = []
+ for app_id in list(state["prepared_apps"]):
+ result = self.remove_app_override(app_id)
+ if result.get("success"):
+ removed_apps.append(app_id)
+ else:
+ failures.append(f"{app_id}: {result.get('error')}")
+ if failures:
+ return self._error_response(
+ dict,
+ "; ".join(failures),
+ removed_apps=removed_apps,
removed_branches=[],
- preserved_branches=[],
- ownership_uncertain=False,
)
- if not self.check_flatpak_available():
- raise RuntimeError("Flatpak is not available; plugin-owned extension metadata was preserved")
- removed, failures = [], []
- for branch in sorted(owned):
- try:
- self._remove_extension(branch)
- removed.append(branch)
- except Exception as error:
- failures.append(f"{branch}: {error}")
- remaining = owned - set(removed)
- self._write_owned_branches(remaining)
+ state = self._read_state()
+ removed_branches = []
+ for branch in sorted(self._owned_branches(state)):
+ result = self.uninstall_extension(branch)
+ if result.get("success"):
+ removed_branches.append(branch)
+ else:
+ failures.append(f"{branch}: {result.get('error')}")
if failures:
return self._error_response(
dict,
"; ".join(failures),
- removed_branches=removed,
- preserved_branches=sorted(remaining),
- ownership_uncertain=False,
+ removed_apps=removed_apps,
+ removed_branches=removed_branches,
)
return self._success_response(
dict,
- "Plugin-owned Flatpak extensions removed",
- removed_branches=removed,
- preserved_branches=[],
- ownership_uncertain=False,
+ "Plugin-owned Flatpak state removed",
+ removed_apps=removed_apps,
+ removed_branches=removed_branches,
)
except Exception as error:
- return self._error_response(
- dict,
- str(error),
- removed_branches=[],
- preserved_branches=[],
- ownership_uncertain=True,
- )
+ return self._error_response(dict, str(error), removed_apps=[], removed_branches=[])
diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py
index 071b19b..d20fabf 100644
--- a/py_modules/lsfg_vk/plugin.py
+++ b/py_modules/lsfg_vk/plugin.py
@@ -1,9 +1,10 @@
import os
-from typing import Any, Dict, Optional
+from typing import Any, Dict
import decky
from .configuration import ConfigurationService
+from .flatpak_profile_service import FlatpakProfileService
from .flatpak_service import FlatpakService
from .installation import InstallationService
from .runtime_service import RuntimeService
@@ -21,6 +22,10 @@ class Plugin:
)
self.configuration_service = ConfigurationService(runtime_service=self.runtime_service)
self.flatpak_service = FlatpakService()
+ self.flatpak_profile_service = FlatpakProfileService(
+ self.flatpak_service,
+ self.configuration_service,
+ )
self.wrapper_service = WrapperService()
async def install_lsfg_vk(self):
@@ -30,27 +35,22 @@ class Plugin:
return self.installation_service.check_installation()
async def uninstall_lsfg_vk(self):
+ flatpak = self.flatpak_service.remove_plugin_owned_environment()
+ if not flatpak.get("success"):
+ return {
+ "success": False,
+ "message": "",
+ "error": flatpak.get("error") or "Could not clean up Flatpak support",
+ "removed_files": None,
+ }
+ self.configuration_service.reset_all_flatpak_configs()
return self.installation_service.uninstall()
async def get_game_configs(self):
return self.configuration_service.get_game_configs()
async def get_installed_games(self):
- result = self.steam_service.get_installed_games()
- if not result.get("success"):
- return result
- cache: Dict[str, Dict[str, Any]] = {}
- for game in result.get("games", []):
- transport = game.get("transport", {})
- if transport.get("kind") != "flatpak":
- continue
- app_id = transport.get("flatpakAppId")
- if not app_id:
- continue
- if app_id not in cache:
- cache[app_id] = self.flatpak_service.resolve_app_support(app_id)
- game["flatpakSupport"] = cache[app_id]
- return result
+ return self.steam_service.get_installed_games()
async def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]):
return self.configuration_service.update_game_config(appid, game_name, config)
@@ -68,17 +68,9 @@ class Plugin:
self,
appid: str,
state: Dict[str, Any],
- shortcut_exe: Optional[str] = None,
command_token_added: bool = False,
- transport: Optional[Dict[str, Any]] = None,
):
- return self.wrapper_service.set(
- appid,
- state,
- shortcut_exe,
- command_token_added,
- transport,
- )
+ return self.wrapper_service.set(appid, state, command_token_added)
async def remove_workaround_state(self, appid: str):
return self.wrapper_service.remove(appid)
@@ -107,20 +99,66 @@ class Plugin:
"error": f"Error reading config file: {error}",
}
+ async def get_debug_file_contents(self):
+ files = (
+ ("config", "LSFG-VK configuration", self.configuration_service.config_file_path),
+ ("workarounds", "Per-app workarounds", self.wrapper_service.sidecar_path),
+ ("wrapper", "Generated launch wrapper", self.wrapper_service.wrapper_path),
+ ("flatpak", "Flatpak ownership state", self.flatpak_service.ownership_path),
+ )
+ contents = []
+ for file_id, label, path in files:
+ item = {
+ "id": file_id,
+ "label": label,
+ "path": str(path),
+ "exists": False,
+ "content": None,
+ "error": None,
+ }
+ try:
+ if path.is_symlink():
+ item["error"] = "Path is a symlink; refusing to read it"
+ elif not path.exists():
+ item["error"] = "File does not exist"
+ elif not path.is_file():
+ item["error"] = "Path is not a regular file"
+ else:
+ item["exists"] = True
+ item["content"] = path.read_text(encoding="utf-8")
+ except Exception as error:
+ item["error"] = f"Error reading file: {error}"
+ contents.append(item)
+ return {
+ "success": True,
+ "message": "Debug file contents retrieved",
+ "error": None,
+ "files": contents,
+ }
+
async def get_lossless_scaling_branch_status(self):
return self.steam_service.get_branch_status()
- async def get_flatpak_support_status(self):
- return self.flatpak_service.get_flatpak_support_status()
+ async def get_flatpak_apps(self):
+ return self.flatpak_profile_service.get_apps()
+
+ async def enable_flatpak_app(self, flatpak_app_id: str):
+ return self.flatpak_profile_service.enable_app(flatpak_app_id)
- async def ensure_flatpak_support(self, flatpak_app_id: str):
- return self.flatpak_service.ensure_app_support(flatpak_app_id)
+ async def update_flatpak_config(self, flatpak_app_id: str, config: Dict[str, Any]):
+ return self.flatpak_profile_service.update_config(flatpak_app_id, config)
- async def repair_flatpak_support(self, flatpak_app_id: str):
- return self.flatpak_service.ensure_app_support(flatpak_app_id)
+ async def get_flatpak_workaround_state(self, flatpak_app_id: str):
+ return self.flatpak_profile_service.get_workaround_state(flatpak_app_id)
- async def set_flatpak_extension_enabled(self, version: str, enabled: bool):
- return self.flatpak_service.set_extension_enabled(version, enabled)
+ async def set_flatpak_workaround_state(self, flatpak_app_id: str, state: Dict[str, Any]):
+ return self.flatpak_profile_service.set_workaround_state(flatpak_app_id, state)
+
+ async def remove_flatpak_app(self, flatpak_app_id: str):
+ return self.flatpak_profile_service.remove_app(flatpak_app_id)
+
+ async def get_running_flatpak_apps(self):
+ return self.flatpak_profile_service.get_running_apps()
async def _main(self):
repair = self.wrapper_service.repair()
@@ -133,13 +171,15 @@ class Plugin:
async def _uninstall(self):
decky.logger.info("decky-lsfg-vk plugin being uninstalled")
- self.installation_service.cleanup_on_uninstall()
try:
- result = self.flatpak_service.remove_plugin_owned_extensions()
- if not result.get("success"):
+ result = self.flatpak_service.remove_plugin_owned_environment()
+ if result.get("success"):
+ self.configuration_service.reset_all_flatpak_configs()
+ else:
decky.logger.warning(result.get("error"))
except Exception as error:
decky.logger.error(f"Error during Flatpak cleanup: {error}")
+ self.installation_service.cleanup_on_uninstall()
decky.logger.info("decky-lsfg-vk plugin uninstall cleanup completed")
async def _migration(self):
diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py
index a952fcb..2108071 100644
--- a/py_modules/lsfg_vk/steam_service.py
+++ b/py_modules/lsfg_vk/steam_service.py
@@ -1,5 +1,4 @@
import re
-import shlex
from pathlib import Path
from typing import Dict, Optional, Tuple
@@ -7,55 +6,8 @@ from .base_service import BaseService
from .constants import (
STEAM_LOSSLESS_SCALING_APP_ID,
STEAM_LOSSLESS_SCALING_BRANCH,
- WRAPPER_FILENAME,
)
-_FLATPAK_APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$")
-_WRAPPER_TOKEN = f"~/{WRAPPER_FILENAME}"
-
-
-def _split_command(value: Optional[str]) -> Optional[list[str]]:
- if not isinstance(value, str) or not value.strip():
- return []
- try:
- return shlex.split(value, posix=True)
- except ValueError:
- return None
-
-
-def _is_managed_wrapper(value: str) -> bool:
- if value in {_WRAPPER_TOKEN, f"$HOME/{WRAPPER_FILENAME}"}:
- return True
- path = Path(value)
- return path.is_absolute() and path.name == WRAPPER_FILENAME
-
-
-def classify_shortcut_transport(executable: Optional[str], launch_options: Optional[str] = None) -> Dict[str, object]:
- executable_tokens = _split_command(executable)
- option_tokens = _split_command(launch_options)
- if executable_tokens is None or option_tokens is None or not executable_tokens:
- return {"kind": "host"}
- direct_flatpak = executable_tokens[0] in {"flatpak", "/usr/bin/flatpak"}
- managed_wrapper = len(executable_tokens) == 1 and _is_managed_wrapper(executable_tokens[0])
- if not direct_flatpak and not managed_wrapper:
- return {"kind": "host"}
- arguments = [*executable_tokens[1:], *option_tokens]
- if not arguments or arguments[0] != "run":
- return {"kind": "host"}
- for argument in arguments[1:]:
- if argument == "--" or argument.startswith("-"):
- continue
- return (
- {"kind": "flatpak", "flatpakAppId": argument}
- if _FLATPAK_APP_ID.fullmatch(argument)
- else {"kind": "host"}
- )
- return {"kind": "host"}
-
-
-def _first_string(values: Dict[str, object], *keys: str) -> Optional[str]:
- return next((values[key] for key in keys if isinstance(values.get(key), str)), None)
-
class SteamService(BaseService):
DEFAULT_BRANCH = "public"
@@ -147,19 +99,11 @@ class SteamService(BaseService):
name = shortcut.get("AppName") or shortcut.get("appname")
if not isinstance(appid, int) or appid == 0 or not isinstance(name, str) or not name:
return None
- executable = _first_string(shortcut, "Exe", "exe", "executable")
- arguments = _first_string(shortcut, "LaunchOptions", "launchoptions", "launch_options", "arguments")
- start_dir = _first_string(shortcut, "StartDir", "startdir", "start_dir")
- game: Dict[str, object] = {
+ return {
"appid": str(appid & 0xFFFFFFFF),
"name": name,
"nonSteam": True,
- "transport": classify_shortcut_transport(executable, arguments),
}
- for key, value in (("executable", executable), ("arguments", arguments), ("startDir", start_dir)):
- if value is not None:
- game[key] = value
- return game
def _shortcut_games(self):
games = {}
@@ -309,7 +253,6 @@ class SteamService(BaseService):
"appid": appid,
"name": self._section_value(content, "AppState", "name") or f"App {appid}",
"nonSteam": False,
- "transport": {"kind": "host"},
}
for game in self._shortcut_games():
games.setdefault(str(game["appid"]), game)
diff --git a/py_modules/lsfg_vk/wrapper_service.py b/py_modules/lsfg_vk/wrapper_service.py
index 1ed39bc..ebe9526 100644
--- a/py_modules/lsfg_vk/wrapper_service.py
+++ b/py_modules/lsfg_vk/wrapper_service.py
@@ -1,12 +1,9 @@
-"""Own the small per-AppID workaround dispatcher used by Steam launches."""
-
from __future__ import annotations
import json
import re
import shlex
import threading
-from pathlib import Path
from typing import Any, Dict, Optional, Tuple
from .base_service import BaseService
@@ -14,8 +11,6 @@ from .constants import WRAPPER_FILENAME
class WrapperService(BaseService):
- """Persist workaround state and compile it into a safe POSIX wrapper."""
-
LEGACY_FORMAT_VERSION = 1
FORMAT_VERSION = 2
LEGACY_MARKER = "# lsfg-vk-wrapper-format: 1"
@@ -35,6 +30,8 @@ class WrapperService(BaseService):
"DISABLE_GAMESCOPE_WSI",
"DXVK_HDR",
"SteamDeck",
+ "DISABLE_LSFGVK",
+ "DISABLE_LSFG",
"DISABLE_VKBASALT",
"ENABLE_VKBASALT",
"MESA_LOADER_DRIVER_OVERRIDE",
@@ -84,51 +81,15 @@ class WrapperService(BaseService):
return state
@classmethod
- def _validate_transport(cls, raw: Any) -> Dict[str, Any]:
- if raw is None:
- return {"kind": "host"}
- if not isinstance(raw, dict):
- raise ValueError("Workaround transport must be an object")
- kind = raw.get("kind")
- if kind == "host":
- return {"kind": "host"}
- if kind == "flatpak":
- app_id = raw.get("flatpakAppId")
- if (
- not isinstance(app_id, str)
- or not re.fullmatch(
- r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$",
- app_id,
- )
- ):
- raise ValueError("Flatpak transport requires a valid application ID")
- return {"kind": "flatpak", "flatpakAppId": app_id}
- raise ValueError("Workaround transport must be host or flatpak")
-
- @classmethod
def _validate_entry(cls, raw: Any) -> Dict[str, Any]:
if not isinstance(raw, dict):
raise ValueError("Workaround AppID entry must be an object")
entry = {
"state": cls._validate_state(raw.get("state")),
"command_token_added": raw.get("command_token_added", False),
- # Version 1 entries had no transport field. They are preserved as
- # host entries until the shortcut is explicitly repaired with the
- # backend's classified transport.
- "transport": cls._validate_transport(raw.get("transport")),
}
if type(entry["command_token_added"]) is not bool:
raise ValueError("command_token_added must be a boolean")
- if entry["transport"]["kind"] == "flatpak" and "shortcut_exe" in raw and raw["shortcut_exe"] is not None:
- shortcut_exe = raw["shortcut_exe"]
- if (
- not isinstance(shortcut_exe, str)
- or not shortcut_exe.startswith("/")
- or "\x00" in shortcut_exe
- or not shortcut_exe.strip()
- ):
- raise ValueError("shortcut_exe must be an absolute executable path")
- entry["shortcut_exe"] = shortcut_exe
return entry
@classmethod
@@ -158,11 +119,11 @@ class WrapperService(BaseService):
if self.sidecar_path.is_symlink() or not self.sidecar_path.is_file():
raise RuntimeError("Workaround state path is not a regular file")
try:
- raw = json.loads(self.sidecar_path.read_text(encoding="utf-8"))
+ content = self.sidecar_path.read_text(encoding="utf-8")
+ raw = json.loads(content)
except (OSError, json.JSONDecodeError) as error:
raise RuntimeError(f"Could not read workaround state: {error}") from error
- document = self._validate_document(raw)
- return document, True, self.sidecar_path.read_text(encoding="utf-8")
+ return self._validate_document(raw), True, content
def _wrapper_marker(self) -> bool:
if self.wrapper_path.is_symlink() or not self.wrapper_path.exists():
@@ -179,29 +140,21 @@ class WrapperService(BaseService):
if not self.wrapper_path.exists() and not self.wrapper_path.is_symlink():
return False
if self.wrapper_path.is_symlink() or not self._wrapper_marker():
- raise RuntimeError(
- f"Refusing to replace unowned wrapper at {self.wrapper_path}"
- )
+ raise RuntimeError(f"Refusing to replace unowned wrapper at {self.wrapper_path}")
return True
@staticmethod
def _shell(value: str) -> str:
return shlex.quote(value)
- @staticmethod
- def _direct_flatpak_tokens(value: str) -> Optional[list[str]]:
- """Parse the supported full executable form: /usr/bin/flatpak run APP."""
- try:
- tokens = shlex.split(value, posix=True)
- except ValueError:
- return None
- if len(tokens) >= 3 and Path(tokens[0]).name == "flatpak" and tokens[1] == "run":
- return tokens
- return None
-
- @classmethod
- def _state_lines(cls, state: Dict[str, Any], shortcut_exe: Optional[str]) -> list[str]:
- lines = [" unset " + " ".join(cls.MANAGED_ENV_KEYS)]
+ def _state_lines(self, state: Dict[str, Any]) -> list[str]:
+ lines = [" unset " + " ".join(self.MANAGED_ENV_KEYS)]
+ lines.extend([
+ ' SteamAppId="$appid"',
+ " export SteamAppId",
+ f" LSFGVK_CONFIG={self._shell(str(self.config_file_path))}",
+ " export LSFGVK_CONFIG",
+ ])
if state["disableGamescopeWsi"]:
lines.extend([" ENABLE_GAMESCOPE_WSI=0", " export ENABLE_GAMESCOPE_WSI"])
if state["disableHdr"]:
@@ -233,72 +186,12 @@ class WrapperService(BaseService):
" fi",
" export DXVK_CONFIG",
])
- lines.append(f" shortcut_exe={cls._shell(shortcut_exe or '')}")
return lines
- def _dll_directory(self) -> Path:
- if self.config_file_path.exists():
- try:
- content = self.config_file_path.read_text(encoding="utf-8")
- match = re.search(
- r'(?m)^[ \t]*dll[ \t]*=[ \t]*"((?:\\.|[^"\\])*)"',
- content,
- )
- if match:
- configured_dll = json.loads('"' + match.group(1) + '"')
- if configured_dll:
- return Path(configured_dll).parent
- except Exception:
- pass
- return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling"
-
- def _flatpak_args(self, state: Dict[str, Any]) -> list[str]:
- config_dir = str(self.config_dir)
- config_file = str(self.config_file_path)
- dll_dir = str(self._dll_directory())
- args = [
- self._shell(f"--filesystem={config_dir}:rw"),
- self._shell(f"--filesystem={dll_dir}:ro"),
- self._shell(f"--env=LSFGVK_CONFIG={config_file}"),
- '"--env=LSFGVK_FLATPAK=1"',
- '"--env=SteamAppId=$appid"',
- '"--unset-env=DISABLE_GAMESCOPE_WSI"',
- '"--unset-env=ENABLE_GAMESCOPE_WSI"' if not state["disableGamescopeWsi"] else
- '"--env=ENABLE_GAMESCOPE_WSI=0"',
- '"--unset-env=DXVK_HDR"' if not state["disableHdr"] else
- '"--env=DXVK_HDR=0"',
- '"--unset-env=SteamDeck"' if not state["disableSteamdeckMode"] else
- '"--env=SteamDeck=0"',
- '"--unset-env=DISABLE_VKBASALT" "--unset-env=ENABLE_VKBASALT"',
- ]
- if state["disableVkbasalt"]:
- args.append('"--env=DISABLE_VKBASALT=1"')
- args.extend([
- '"--unset-env=MESA_LOADER_DRIVER_OVERRIDE" "--unset-env=__GLX_VENDOR_LIBRARY_NAME" "--unset-env=GALLIUM_DRIVER"',
- ])
- if state["enableZink"]:
- args.extend([
- '"--env=__GLX_VENDOR_LIBRARY_NAME=mesa"',
- '"--env=MESA_LOADER_DRIVER_OVERRIDE=zink"',
- '"--env=GALLIUM_DRIVER=zink"',
- ])
- args.extend([
- '"--unset-env=DXVK_FRAME_RATE"',
- ])
- static_args = " ".join(args)
- return [
- ' if [ -n "${DXVK_CONFIG+x}" ]; then',
- f' set -- "$flatpak_command" {static_args} "--env=DXVK_CONFIG=$DXVK_CONFIG" "$@"',
- " else",
- f' set -- "$flatpak_command" {static_args} "$@"',
- " fi",
- ]
-
def _render_wrapper(self, document: Dict[str, Any]) -> str:
lines = [
"#!/bin/sh",
self.MARKER,
- "# Generated by Decky LSFG-VK; edits will be rejected on the next update.",
"",
"appid=",
'case "${SteamAppId-}" in',
@@ -317,77 +210,25 @@ class WrapperService(BaseService):
' *) appid="${STEAM_COMPAT_APP_ID}" ;;',
" esac",
"fi",
- "shortcut_exe=",
'case "$appid" in',
]
for appid in sorted(document["apps"], key=lambda value: int(value)):
- entry = document["apps"][appid]
lines.append(f" {appid})")
- lines.extend(self._state_lines(entry["state"], entry.get("shortcut_exe")))
+ lines.extend(self._state_lines(document["apps"][appid]["state"]))
lines.append(" ;;")
lines.extend([
"esac",
- "",
- 'if [ -n "$shortcut_exe" ]; then',
- ])
- # The arguments are emitted per branch below so the values are static and
- # the wrapper never needs a JSON parser or another helper executable.
- lines.append(' case "$appid" in')
- for appid in sorted(document["apps"], key=lambda value: int(value)):
- entry = document["apps"][appid]
- transport = entry.get("transport", {"kind": "host"})
- if transport.get("kind") != "flatpak":
- continue
- shortcut_exe = entry.get("shortcut_exe", "")
- direct_flatpak_tokens = self._direct_flatpak_tokens(shortcut_exe)
- if direct_flatpak_tokens is None and Path(shortcut_exe).name != "flatpak":
- raise ValueError(
- f"Flatpak target {appid} does not use a direct flatpak executable"
- )
- lines.append(f" {appid})")
- lines.extend([
- *(
- [
- f" shortcut_exe={self._shell(direct_flatpak_tokens[0])}",
- " set -- "
- + " ".join(self._shell(token) for token in direct_flatpak_tokens[1:])
- + ' "$@"',
- ]
- if direct_flatpak_tokens
- else []
- ),
- ' if [ "${1-}" != "run" ]; then',
- ' echo "lsfg-vk: Flatpak shortcut must use direct flatpak run transport" >&2',
- " exit 64",
- " fi",
- ' flatpak_command="$1"',
- " shift",
- " flatpak_target=",
- ' for flatpak_arg in "$@"; do',
- ' case "$flatpak_arg" in',
- ' -*) ;;',
- ' *) flatpak_target="$flatpak_arg"; break ;;',
- " esac",
- " done",
- f' if [ "$flatpak_target" != {self._shell(transport["flatpakAppId"])} ]; then',
- ' echo "lsfg-vk: Flatpak shortcut application ID changed externally" >&2',
- " exit 64",
- " fi",
- ])
- lines.extend(self._flatpak_args(entry["state"]))
- lines.append(" ;;")
- lines.extend([
- " esac",
- ' exec "$shortcut_exe" "$@"',
- "fi",
'exec "$@"',
"",
])
return "\n".join(lines)
def _write_document(self, document: Dict[str, Any]) -> None:
- content = json.dumps(document, indent=2, sort_keys=True) + "\n"
- self._write_file(self.sidecar_path, content, 0o644)
+ self._write_file(
+ self.sidecar_path,
+ json.dumps(document, indent=2, sort_keys=True) + "\n",
+ 0o644,
+ )
def _write_pair(self, document: Dict[str, Any]) -> None:
old_sidecar_exists = self.sidecar_path.exists()
@@ -421,9 +262,7 @@ class WrapperService(BaseService):
"state": dict(entry["state"]) if entry else None,
"wrapper_path": self.WRAPPER_TOKEN,
"wrapper_owned": self._wrapper_marker() if document["apps"] else False,
- "shortcut_exe": entry.get("shortcut_exe") if entry else None,
"command_token_added": entry.get("command_token_added", False) if entry else False,
- "transport": dict(entry.get("transport", {"kind": "host"})) if entry else None,
}
def get(self, appid: str) -> Dict[str, Any]:
@@ -448,9 +287,7 @@ class WrapperService(BaseService):
self,
appid: str,
state: Dict[str, Any],
- shortcut_exe: Optional[str] = None,
command_token_added: bool = False,
- transport: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
try:
normalized = self._valid_appid(appid)
@@ -460,27 +297,10 @@ class WrapperService(BaseService):
with self._lock:
self._assert_wrapper_owned_or_absent()
document, _, _ = self._read_document()
- previous_entry = document["apps"].get(normalized)
- selected_transport = self._validate_transport(
- transport
- if transport is not None
- else (
- previous_entry.get("transport")
- if previous_entry
- else None
- )
- )
- entry: Dict[str, Any] = {
+ document["apps"][normalized] = {
"state": validated_state,
- "command_token_added": bool(command_token_added),
- "transport": selected_transport,
+ "command_token_added": command_token_added,
}
- if selected_transport["kind"] == "flatpak":
- if shortcut_exe is not None:
- entry = self._validate_entry({**entry, "shortcut_exe": shortcut_exe})
- elif previous_entry and "shortcut_exe" in previous_entry:
- entry["shortcut_exe"] = previous_entry["shortcut_exe"]
- document["apps"][normalized] = entry
self._write_pair(document)
return self._response(document, normalized)
except Exception as error:
@@ -517,7 +337,6 @@ class WrapperService(BaseService):
}
def repair(self) -> Dict[str, Any]:
- """Regenerate a missing owned wrapper without importing old global state."""
try:
with self._lock:
document, _, _ = self._read_document()
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-11 08:32:36 +0000