874 lines
41 KiB
Python
Executable File
874 lines
41 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
"""Redacted diagnostics and bounded repairs for Panama-owned functionality."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ctypes
|
|
import errno
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import signal
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from types import MappingProxyType
|
|
from typing import Callable, Literal
|
|
|
|
Status = Literal["ok", "warning", "error", "unconfigured"]
|
|
Group = Literal["desktop-foundation", "input-media", "integrations", "panama-tools"]
|
|
ActionKind = Literal["repair", "open", "instructions"]
|
|
InhibitorRow = tuple[str, str, str, str, str, str, str, str]
|
|
|
|
AT_FDCWD = -100
|
|
RENAME_NOREPLACE = 1
|
|
RENAME_EXCHANGE = 2
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Action:
|
|
kind: ActionKind
|
|
label: str
|
|
confirm: bool = False
|
|
# Only authored Settings page IDs and instruction IDs are allowed here.
|
|
# Repair commands never receive a caller-controlled target.
|
|
target: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Check:
|
|
id: str
|
|
group: Group
|
|
title: str
|
|
status: Status
|
|
detail: str
|
|
action: Action | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DoctorConfig:
|
|
root: Path
|
|
home: Path
|
|
config_home: Path
|
|
state_home: Path
|
|
runtime_dir: Path
|
|
path: str
|
|
timeout: float
|
|
|
|
@property
|
|
def command_env(self) -> dict[str, str]:
|
|
environment = {
|
|
"PATH": self.path,
|
|
"HOME": str(self.home),
|
|
"XDG_CONFIG_HOME": str(self.config_home),
|
|
"XDG_STATE_HOME": str(self.state_home),
|
|
"XDG_RUNTIME_DIR": str(self.runtime_dir),
|
|
}
|
|
for name in PROBE_ENVIRONMENT_KEYS:
|
|
if value := os.environ.get(name):
|
|
environment[name] = value
|
|
return environment
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CommandResult:
|
|
state: Literal["ok", "missing", "timeout", "failed", "unavailable"]
|
|
stdout: str = ""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RepairResult:
|
|
check_id: str
|
|
accepted: bool
|
|
exit_code: int
|
|
message: str
|
|
|
|
def as_json(self) -> dict[str, object]:
|
|
return {
|
|
"schemaVersion": 1,
|
|
"checkId": self.check_id,
|
|
"accepted": self.accepted,
|
|
"exitCode": self.exit_code,
|
|
"message": self.message,
|
|
}
|
|
|
|
|
|
CHECK_ORDER = (
|
|
"desktop.hyprland", "desktop.quickshell", "desktop.notifications", "desktop.portals",
|
|
"desktop.hyprpaper", "desktop.hypridle", "desktop.vicinae", "input.pipewire",
|
|
"input.clipboard", "input.wallpaper", "input.capture", "input.ocr", "input.brightness",
|
|
"integration.nextcloud", "integration.rustdesk", "integration.kdeconnect", "integration.bluebubbles",
|
|
"integration.home-assistant", "integration.calendar", "panama.runtime-links", "panama.vicinae-commands",
|
|
"panama.selected-terminal", "panama.selected-launcher", "panama.processes", "panama.caffeine",
|
|
)
|
|
|
|
SYSTEMCTL_COMMANDS = {
|
|
"hyprpaper": ("systemctl", "--user", "is-active", "--quiet", "hyprpaper.service"),
|
|
"hypridle": ("systemctl", "--user", "is-active", "--quiet", "hypridle.service"),
|
|
"vicinae": ("systemctl", "--user", "is-active", "--quiet", "vicinae.service"),
|
|
"pipewire": ("systemctl", "--user", "is-active", "--quiet", "pipewire.service"),
|
|
"nextcloud": ("systemctl", "--user", "is-active", "--quiet", "nextcloud.service"),
|
|
"rustdesk": ("systemctl", "--user", "is-active", "--quiet", "rustdesk.service"),
|
|
}
|
|
REPAIR_COMMANDS = MappingProxyType({
|
|
"desktop.hyprpaper": ("systemctl", "--user", "restart", "hyprpaper.service"),
|
|
"desktop.hypridle": ("systemctl", "--user", "restart", "hypridle.service"),
|
|
"desktop.vicinae": ("systemctl", "--user", "restart", "vicinae.service"),
|
|
"desktop.quickshell": ("panama-action", "restart-shell"),
|
|
})
|
|
RUNTIME_LINK_TARGETS = (
|
|
("hypr", Path("config/dot/hypr")),
|
|
("quickshell", Path("config/dot/quickshell")),
|
|
("uwsm", Path("config/dot/uwsm")),
|
|
("vicinae", Path("config/dot/vicinae")),
|
|
)
|
|
REPAIR_IDS = frozenset((*REPAIR_COMMANDS.keys(), "panama.runtime-links", "panama.vicinae-commands", "panama.caffeine"))
|
|
PROCESS_NAMES = ("vicinae", "hyprpaper", "hypridle")
|
|
VERSION_PATTERN = re.compile(r"\b\d+(?:\.\d+){0,3}(?:[-+._][A-Za-z0-9._-]+)?\b")
|
|
REVISION_PATTERN = re.compile(r"\b[0-9a-f]{7,40}\b", re.IGNORECASE)
|
|
PROBE_ENVIRONMENT_KEYS = (
|
|
"LANG",
|
|
"LC_ALL",
|
|
"LC_CTYPE",
|
|
"TZ",
|
|
"DBUS_SESSION_BUS_ADDRESS",
|
|
"WAYLAND_DISPLAY",
|
|
"DISPLAY",
|
|
"XAUTHORITY",
|
|
"PANAMA_DOCTOR_FIXTURE_STOPPED",
|
|
"PANAMA_DOCTOR_FIXTURE_PROCESSES",
|
|
"PANAMA_DOCTOR_FIXTURE_BUS",
|
|
"PANAMA_DOCTOR_FIXTURE_QS",
|
|
"PANAMA_DOCTOR_FIXTURE_QS_VERSION",
|
|
"PANAMA_DOCTOR_FIXTURE_BLUEBUBBLES",
|
|
"PANAMA_DOCTOR_FIXTURE_CALENDAR",
|
|
"PANAMA_DOCTOR_FIXTURE_BRIGHTNESS",
|
|
"PANAMA_DOCTOR_FIXTURE_CAFFEINE",
|
|
)
|
|
CHECK_TITLES = {
|
|
"desktop.hyprland": "Hyprland",
|
|
"desktop.quickshell": "Quickshell",
|
|
"desktop.notifications": "Notifications",
|
|
"desktop.portals": "Desktop portals",
|
|
"desktop.hyprpaper": "Hyprpaper",
|
|
"desktop.hypridle": "Hypridle",
|
|
"desktop.vicinae": "Vicinae",
|
|
"input.pipewire": "PipeWire",
|
|
"input.clipboard": "Clipboard",
|
|
"input.wallpaper": "Wallpaper",
|
|
"input.capture": "Capture",
|
|
"input.ocr": "OCR",
|
|
"input.brightness": "External monitor brightness",
|
|
"integration.nextcloud": "Nextcloud",
|
|
"integration.rustdesk": "RustDesk",
|
|
"integration.kdeconnect": "KDE Connect",
|
|
"integration.bluebubbles": "BlueBubbles",
|
|
"integration.home-assistant": "Home Assistant",
|
|
"integration.calendar": "Calendar",
|
|
"panama.runtime-links": "Panama runtime links",
|
|
"panama.vicinae-commands": "Panama commands",
|
|
"panama.selected-terminal": "Selected terminal",
|
|
"panama.selected-launcher": "Selected launcher",
|
|
"panama.processes": "Panama processes",
|
|
"panama.caffeine": "Caffeine inhibitor",
|
|
}
|
|
|
|
|
|
def environment_path(name: str, default: Path) -> Path:
|
|
value = os.environ.get(name)
|
|
return Path(value).expanduser() if value else default
|
|
|
|
|
|
def config_from_environment() -> DoctorConfig:
|
|
home = environment_path("PANAMA_DOCTOR_HOME", Path.home())
|
|
config_home = environment_path("PANAMA_DOCTOR_CONFIG_HOME", Path(os.environ.get("XDG_CONFIG_HOME", home / ".config")))
|
|
state_home = environment_path("PANAMA_DOCTOR_STATE_HOME", Path(os.environ.get("XDG_STATE_HOME", home / ".local/state")))
|
|
runtime_dir = environment_path("PANAMA_DOCTOR_RUNTIME_DIR", Path(os.environ.get("XDG_RUNTIME_DIR", "/run/user/0")))
|
|
root = environment_path("PANAMA_DOCTOR_ROOT", Path(__file__).resolve().parents[4])
|
|
try:
|
|
timeout = float(os.environ.get("PANAMA_DOCTOR_TIMEOUT", "3"))
|
|
except ValueError:
|
|
timeout = 3.0
|
|
return DoctorConfig(root, home, config_home, state_home, runtime_dir, os.environ.get("PANAMA_DOCTOR_PATH", os.environ.get("PATH", "")), max(0.05, min(timeout, 15.0)))
|
|
|
|
|
|
def run_command(command: tuple[str, ...], config: DoctorConfig, cwd: Path | None = None) -> CommandResult:
|
|
"""Run an authored read-only command without reporting its unparsed output."""
|
|
try:
|
|
completed = subprocess.run(command, capture_output=True, text=True, timeout=config.timeout, check=False, env=config.command_env, cwd=cwd)
|
|
except FileNotFoundError:
|
|
return CommandResult("missing")
|
|
except subprocess.TimeoutExpired:
|
|
return CommandResult("timeout")
|
|
except OSError:
|
|
return CommandResult("unavailable")
|
|
if completed.returncode != 0:
|
|
return CommandResult("failed")
|
|
return CommandResult("ok", completed.stdout)
|
|
|
|
|
|
def run_repair_command(command: tuple[str, ...], config: DoctorConfig, cwd: Path | None = None) -> tuple[int, str]:
|
|
"""Execute one authored repair argv and retain output only for strict parsing."""
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=config.timeout,
|
|
check=False,
|
|
env=config.command_env,
|
|
cwd=cwd,
|
|
)
|
|
except FileNotFoundError:
|
|
return 127, ""
|
|
except subprocess.TimeoutExpired:
|
|
return 124, ""
|
|
except OSError:
|
|
return 126, ""
|
|
exit_code = completed.returncode if 0 <= completed.returncode <= 255 else 1
|
|
return exit_code, completed.stdout
|
|
|
|
|
|
def executable_exists(name: str, config: DoctorConfig) -> bool:
|
|
return shutil.which(name, path=config.path) is not None
|
|
|
|
|
|
def action_json(action: Action) -> dict[str, object]:
|
|
result: dict[str, object] = {"kind": action.kind, "label": action.label, "confirm": action.confirm}
|
|
if action.target is not None:
|
|
result["target"] = action.target
|
|
return result
|
|
|
|
|
|
def check_json(check: Check) -> dict[str, object]:
|
|
result: dict[str, object] = {"id": check.id, "group": check.group, "title": check.title, "status": check.status, "detail": check.detail}
|
|
if check.action is not None:
|
|
result["action"] = action_json(check.action)
|
|
return result
|
|
|
|
|
|
def group_for(check_id: str) -> Group:
|
|
if check_id.startswith("desktop."):
|
|
return "desktop-foundation"
|
|
if check_id.startswith("input."):
|
|
return "input-media"
|
|
if check_id.startswith("integration."):
|
|
return "integrations"
|
|
return "panama-tools"
|
|
|
|
|
|
def service_check(check_id: str, title: str, service: str, config: DoctorConfig, action: Action | None = None) -> Check:
|
|
result = run_command(SYSTEMCTL_COMMANDS[service], config)
|
|
if result.state == "ok":
|
|
return Check(check_id, group_for(check_id), title, "ok", "Service is active.")
|
|
if result.state in {"missing", "unavailable"}:
|
|
return Check(check_id, group_for(check_id), title, "error", "Required system service probe is unavailable.")
|
|
return Check(check_id, group_for(check_id), title, "warning", "Service is not active.", action)
|
|
|
|
|
|
def ipc_target(config: DoctorConfig, target: str) -> CommandResult:
|
|
result = run_command(("qs", "ipc", "show"), config)
|
|
if result.state != "ok":
|
|
return result
|
|
return CommandResult("ok") if f"target {target}" in result.stdout.splitlines() else CommandResult("failed")
|
|
|
|
|
|
def simple_ipc_check(check_id: str, title: str, target: str, config: DoctorConfig) -> Check:
|
|
result = ipc_target(config, target)
|
|
if result.state == "ok":
|
|
return Check(check_id, "input-media", title, "ok", "Panama IPC target is available.")
|
|
if result.state == "missing":
|
|
return Check(check_id, "input-media", title, "error", "Required Quickshell executable is unavailable.")
|
|
if result.state == "timeout":
|
|
return Check(check_id, "input-media", title, "warning", "Panama IPC probe timed out.")
|
|
return Check(check_id, "input-media", title, "warning", "Panama IPC target is unavailable.")
|
|
|
|
|
|
def check_hyprland(config: DoctorConfig) -> Check:
|
|
if "hyprland" in os.environ.get("XDG_CURRENT_DESKTOP", "").casefold():
|
|
return Check("desktop.hyprland", "desktop-foundation", "Hyprland", "ok", "Hyprland session detected.")
|
|
return Check("desktop.hyprland", "desktop-foundation", "Hyprland", "error", "Hyprland session is not active.")
|
|
|
|
|
|
def check_quickshell(config: DoctorConfig) -> Check:
|
|
result = run_command(("qs", "--version"), config)
|
|
repair = Action("repair", "Restart Panama", True)
|
|
if result.state == "ok" and VERSION_PATTERN.search(result.stdout):
|
|
return Check("desktop.quickshell", "desktop-foundation", "Quickshell", "ok", "Quickshell executable is available.")
|
|
if result.state == "missing":
|
|
return Check("desktop.quickshell", "desktop-foundation", "Quickshell", "error", "Required Quickshell executable is unavailable.", repair)
|
|
return Check("desktop.quickshell", "desktop-foundation", "Quickshell", "warning", "Quickshell probe returned an invalid result.", repair)
|
|
|
|
|
|
def check_notifications(config: DoctorConfig) -> Check:
|
|
result = ipc_target(config, "notifications")
|
|
return Check("desktop.notifications", "desktop-foundation", "Notifications", "ok", "Notification service is available.") if result.state == "ok" else Check("desktop.notifications", "desktop-foundation", "Notifications", "warning", "Notification service is unavailable.")
|
|
|
|
|
|
def check_portals(config: DoctorConfig) -> Check:
|
|
result = run_command(("busctl", "--user", "--no-pager", "list"), config)
|
|
if result.state == "ok" and any(line.startswith("org.freedesktop.portal.Desktop ") for line in result.stdout.splitlines()):
|
|
return Check("desktop.portals", "desktop-foundation", "Desktop portals", "ok", "Desktop portal service is available.")
|
|
detail = "Desktop portal probe timed out." if result.state == "timeout" else "Desktop portal probe is unavailable." if result.state == "missing" else "Desktop portal service is unavailable."
|
|
return Check("desktop.portals", "desktop-foundation", "Desktop portals", "warning", detail)
|
|
|
|
|
|
def check_brightness(config: DoctorConfig) -> Check:
|
|
result = run_command(("panama-brightness", "list"), config)
|
|
instructions = Action("instructions", "View setup instructions", target="ddc-permissions")
|
|
if result.state == "timeout":
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "warning", "DDC/CI probe timed out.", instructions)
|
|
if result.state != "ok":
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "warning", "DDC/CI support is unavailable.", instructions)
|
|
try:
|
|
listing = json.loads(result.stdout)
|
|
displays, error = listing["displays"], listing["error"]
|
|
if not isinstance(displays, list) or not isinstance(error, str):
|
|
raise ValueError
|
|
except (json.JSONDecodeError, KeyError, TypeError, ValueError):
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "warning", "DDC/CI probe returned an invalid result.", instructions)
|
|
if error:
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "warning", "No accessible DDC/CI bus.", instructions)
|
|
if not displays:
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "unconfigured", "No DDC/CI display is configured.")
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "ok", f"{len(displays)} DDC/CI display{'s' if len(displays) != 1 else ''} available.")
|
|
|
|
|
|
def check_nextcloud(config: DoctorConfig) -> Check:
|
|
if not (config.config_home / "autostart" / "nextcloud.desktop").is_file():
|
|
return Check("integration.nextcloud", "integrations", "Nextcloud", "unconfigured", "Nextcloud autostart is not configured.")
|
|
return service_check("integration.nextcloud", "Nextcloud", "nextcloud", config, Action("open", "Open Nextcloud"))
|
|
|
|
|
|
def check_rustdesk(config: DoctorConfig) -> Check:
|
|
if not executable_exists("rustdesk", config):
|
|
return Check("integration.rustdesk", "integrations", "RustDesk", "unconfigured", "RustDesk is not installed.")
|
|
return service_check("integration.rustdesk", "RustDesk", "rustdesk", config, Action("open", "Open RustDesk"))
|
|
|
|
|
|
def check_kdeconnect(config: DoctorConfig) -> Check:
|
|
if not executable_exists("kdeconnect-cli", config):
|
|
return Check("integration.kdeconnect", "integrations", "KDE Connect", "unconfigured", "KDE Connect is not installed.")
|
|
result = run_command(("busctl", "--user", "--no-pager", "list"), config)
|
|
if result.state == "ok" and any(line.startswith("org.kde.kdeconnect ") for line in result.stdout.splitlines()):
|
|
return Check("integration.kdeconnect", "integrations", "KDE Connect", "ok", "KDE Connect service is available.")
|
|
return Check("integration.kdeconnect", "integrations", "KDE Connect", "warning", "KDE Connect service is unavailable.", Action("open", "Open KDE Connect"))
|
|
|
|
|
|
def check_bluebubbles(config: DoctorConfig) -> Check:
|
|
result = run_command(("flatpak", "info", "app.bluebubbles.BlueBubbles"), config)
|
|
if result.state == "ok":
|
|
return Check("integration.bluebubbles", "integrations", "BlueBubbles", "ok", "BlueBubbles is installed.")
|
|
if result.state in {"missing", "failed"}:
|
|
return Check("integration.bluebubbles", "integrations", "BlueBubbles", "unconfigured", "BlueBubbles is not installed.")
|
|
return Check("integration.bluebubbles", "integrations", "BlueBubbles", "warning", "BlueBubbles installation probe timed out.", Action("open", "Open BlueBubbles"))
|
|
|
|
|
|
def check_home_assistant(config: DoctorConfig) -> Check:
|
|
configured = all(name in os.environ for name in ("PANAMA_HOME_ASSISTANT_URL", "PANAMA_HOME_ASSISTANT_TOKEN"))
|
|
helper = config.config_home / "quickshell" / "scripts" / "panama-home-assistant"
|
|
if not configured:
|
|
return Check("integration.home-assistant", "integrations", "Home Assistant", "unconfigured", "Home Assistant is not configured.")
|
|
if not helper.is_file():
|
|
return Check("integration.home-assistant", "integrations", "Home Assistant", "warning", "Home Assistant bridge is unavailable.", Action("open", "Open Home settings", target="home-phone"))
|
|
return Check("integration.home-assistant", "integrations", "Home Assistant", "ok", "Home Assistant credentials are configured.")
|
|
|
|
|
|
def check_calendar(config: DoctorConfig) -> Check:
|
|
result = run_command(("calendar-agenda", "probe"), config)
|
|
action = Action("open", "Open Date & Time", target="datetime")
|
|
if result.state == "missing":
|
|
return Check("integration.calendar", "integrations", "Calendar", "unconfigured", "Calendar integration is not installed.")
|
|
if result.state == "timeout":
|
|
return Check("integration.calendar", "integrations", "Calendar", "warning", "Calendar probe timed out.", action)
|
|
if result.state != "ok":
|
|
return Check("integration.calendar", "integrations", "Calendar", "warning", "Calendar probe failed.", action)
|
|
try:
|
|
enabled_sources = json.loads(result.stdout)["enabledSources"]
|
|
if not isinstance(enabled_sources, int) or isinstance(enabled_sources, bool):
|
|
raise ValueError
|
|
except (json.JSONDecodeError, KeyError, TypeError, ValueError):
|
|
return Check("integration.calendar", "integrations", "Calendar", "warning", "Calendar probe returned an invalid result.", action)
|
|
if enabled_sources <= 0:
|
|
return Check("integration.calendar", "integrations", "Calendar", "unconfigured", "No enabled calendar source is configured.")
|
|
return Check("integration.calendar", "integrations", "Calendar", "ok", f"{enabled_sources} enabled calendar source{'s' if enabled_sources != 1 else ''} configured.")
|
|
|
|
|
|
def check_runtime_links(config: DoctorConfig) -> Check:
|
|
def valid_link(name: str, relative_source: Path) -> bool:
|
|
destination = config.config_home / name
|
|
source = config.root / relative_source
|
|
try:
|
|
return source.is_dir() and destination.is_symlink() \
|
|
and destination.resolve(strict=False) == source.resolve(strict=True)
|
|
except OSError:
|
|
return False
|
|
|
|
if any(not valid_link(name, relative_source) for name, relative_source in RUNTIME_LINK_TARGETS):
|
|
return Check("panama.runtime-links", "panama-tools", "Panama runtime links", "warning", "One or more Panama runtime links are unavailable.", Action("repair", "Repair runtime links"))
|
|
return Check("panama.runtime-links", "panama-tools", "Panama runtime links", "ok", "Panama runtime links are available.")
|
|
|
|
|
|
def check_vicinae_commands(config: DoctorConfig) -> Check:
|
|
source = config.root / "config/local/share/vicinae/scripts"
|
|
installed = config.home / ".local/share/vicinae/scripts/panama"
|
|
try:
|
|
linked = source.is_dir() and installed.is_symlink() \
|
|
and installed.resolve(strict=False) == source.resolve(strict=True)
|
|
except OSError:
|
|
linked = False
|
|
if linked:
|
|
return Check("panama.vicinae-commands", "panama-tools", "Panama commands", "ok", "Panama Vicinae commands are linked.")
|
|
return Check("panama.vicinae-commands", "panama-tools", "Panama commands", "warning", "Panama Vicinae commands are not linked.", Action("repair", "Repair command link"))
|
|
|
|
|
|
def executable_check(check_id: str, title: str, executable: str, config: DoctorConfig) -> Check:
|
|
if executable_exists(executable, config):
|
|
return Check(check_id, group_for(check_id), title, "ok", f"{title} executable is available.")
|
|
return Check(check_id, group_for(check_id), title, "warning", f"{title} executable is unavailable.")
|
|
|
|
|
|
def check_processes(config: DoctorConfig) -> Check:
|
|
# `qs` is both the long-running shell and every short-lived IPC client.
|
|
# Counting it with pgrep races the other parallel health probes and reports
|
|
# duplicates whenever one of them happens to call `qs ipc`. The instance
|
|
# list is the authoritative view and contains only actual shells.
|
|
quickshell = run_command(("qs", "list"), config)
|
|
if quickshell.state == "ok":
|
|
quickshell_count = sum(
|
|
line.startswith("Instance ") for line in quickshell.stdout.splitlines()
|
|
)
|
|
elif quickshell.state == "failed":
|
|
quickshell_count = 0
|
|
else:
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "warning", "Process probe is unavailable.")
|
|
|
|
counts: list[int] = [quickshell_count]
|
|
for name in PROCESS_NAMES:
|
|
result = run_command(("pgrep", "-u", str(os.getuid()), "-x", name), config)
|
|
if result.state == "ok":
|
|
pids = result.stdout.splitlines()
|
|
if not pids or any(not pid.isdecimal() for pid in pids):
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "warning", "Process probe returned an invalid result.")
|
|
counts.append(len(pids))
|
|
elif result.state == "failed":
|
|
counts.append(0)
|
|
else:
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "warning", "Process probe is unavailable.")
|
|
if any(count > 1 for count in counts):
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "warning", "Duplicate Panama desktop processes detected.")
|
|
if counts[0] == 0:
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "error", "Quickshell process is not running.")
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "ok", "Panama desktop process counts are normal.")
|
|
|
|
|
|
def check_caffeine(config: DoctorConfig) -> Check:
|
|
result = run_command(("systemd-inhibit", "--list", "--no-pager", "--no-legend"), config)
|
|
if result.state != "ok":
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe is unavailable.")
|
|
inhibitor_rows = parse_caffeine_rows(result.stdout, str(os.getuid()))
|
|
if inhibitor_rows is None:
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe returned an invalid result.")
|
|
inhibitors = len(dict.fromkeys(int(row[3]) for row in inhibitor_rows))
|
|
if inhibitors > 1:
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Duplicate Panama Caffeine inhibitors detected.", Action("repair", "Release duplicate inhibitors"))
|
|
if inhibitors == 1:
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "ok", "One Panama Caffeine inhibitor is active.")
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "ok", "No Panama Caffeine inhibitor is active.")
|
|
|
|
|
|
def parse_version(result: CommandResult, pattern: re.Pattern[str] = VERSION_PATTERN) -> str:
|
|
match = pattern.search(result.stdout) if result.state == "ok" else None
|
|
return match.group(0) if match else "unavailable"
|
|
|
|
|
|
def context_versions(config: DoctorConfig) -> list[dict[str, str]]:
|
|
hyprland = run_command(("hyprctl", "version"), config)
|
|
quickshell = run_command(("qs", "--version"), config)
|
|
revision = run_command(("git", "rev-parse", "--short", "HEAD"), config, config.root)
|
|
fedora = "unavailable"
|
|
try:
|
|
match = re.search(r"^VERSION_ID=\"?([^\n\"]+)", Path("/etc/os-release").read_text(encoding="utf-8"), re.MULTILINE)
|
|
if match and re.fullmatch(r"[0-9.]+", match.group(1)):
|
|
fedora = match.group(1)
|
|
except OSError:
|
|
pass
|
|
return [{"id": "hyprland", "version": parse_version(hyprland)}, {"id": "quickshell", "version": parse_version(quickshell)}, {"id": "fedora", "version": fedora}, {"id": "panama", "version": parse_version(revision, REVISION_PATTERN)}]
|
|
|
|
|
|
def unavailable_check(check_id: str) -> Check:
|
|
return Check(check_id, group_for(check_id), CHECK_TITLES[check_id], "warning", "Diagnostic probe could not be completed.")
|
|
|
|
|
|
def unavailable_versions() -> list[dict[str, str]]:
|
|
return [{"id": name, "version": "unavailable"} for name in ("hyprland", "quickshell", "fedora", "panama")]
|
|
|
|
|
|
def collect_checks(config: DoctorConfig) -> list[Check]:
|
|
probes: dict[str, Callable[[], Check]] = {
|
|
"desktop.hyprland": lambda: check_hyprland(config), "desktop.quickshell": lambda: check_quickshell(config), "desktop.notifications": lambda: check_notifications(config), "desktop.portals": lambda: check_portals(config),
|
|
"desktop.hyprpaper": lambda: service_check("desktop.hyprpaper", "Hyprpaper", "hyprpaper", config, Action("repair", "Restart Hyprpaper")), "desktop.hypridle": lambda: service_check("desktop.hypridle", "Hypridle", "hypridle", config, Action("repair", "Restart Hypridle")), "desktop.vicinae": lambda: service_check("desktop.vicinae", "Vicinae", "vicinae", config, Action("repair", "Restart Vicinae")), "input.pipewire": lambda: service_check("input.pipewire", "PipeWire", "pipewire", config),
|
|
"input.clipboard": lambda: simple_ipc_check("input.clipboard", "Clipboard", "clipboard", config), "input.wallpaper": lambda: simple_ipc_check("input.wallpaper", "Wallpaper", "wallpaper", config), "input.capture": lambda: simple_ipc_check("input.capture", "Capture", "capture", config), "input.ocr": lambda: executable_check("input.ocr", "OCR", "tesseract", config), "input.brightness": lambda: check_brightness(config),
|
|
"integration.nextcloud": lambda: check_nextcloud(config), "integration.rustdesk": lambda: check_rustdesk(config), "integration.kdeconnect": lambda: check_kdeconnect(config), "integration.bluebubbles": lambda: check_bluebubbles(config), "integration.home-assistant": lambda: check_home_assistant(config), "integration.calendar": lambda: check_calendar(config),
|
|
"panama.runtime-links": lambda: check_runtime_links(config), "panama.vicinae-commands": lambda: check_vicinae_commands(config), "panama.selected-terminal": lambda: executable_check("panama.selected-terminal", "Selected terminal", "kitty", config), "panama.selected-launcher": lambda: executable_check("panama.selected-launcher", "Selected launcher", "vicinae", config), "panama.processes": lambda: check_processes(config), "panama.caffeine": lambda: check_caffeine(config),
|
|
}
|
|
with ThreadPoolExecutor(max_workers=8) as executor:
|
|
futures = {check_id: executor.submit(probes[check_id]) for check_id in CHECK_ORDER}
|
|
checks: list[Check] = []
|
|
for check_id in CHECK_ORDER:
|
|
try:
|
|
checks.append(futures[check_id].result())
|
|
except Exception:
|
|
checks.append(unavailable_check(check_id))
|
|
return checks
|
|
|
|
|
|
def snapshot(config: DoctorConfig) -> dict[str, object]:
|
|
try:
|
|
checks = collect_checks(config)
|
|
except Exception:
|
|
checks = [unavailable_check(check_id) for check_id in CHECK_ORDER]
|
|
counts = {status: sum(check.status == status for check in checks) for status in ("ok", "warning", "error", "unconfigured")}
|
|
overall: Literal["healthy", "warning", "error"] = "error" if counts["error"] else "warning" if counts["warning"] else "healthy"
|
|
session = "hyprland" if "hyprland" in os.environ.get("XDG_CURRENT_DESKTOP", "").casefold() else "other"
|
|
try:
|
|
versions = context_versions(config)
|
|
except Exception:
|
|
versions = unavailable_versions()
|
|
return {"schemaVersion": 1, "generatedAt": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"), "summary": {"status": overall, "healthy": counts["ok"], "warnings": counts["warning"], "errors": counts["error"], "unconfigured": counts["unconfigured"]}, "context": {"session": session, "versions": versions}, "checks": [check_json(check) for check in checks]}
|
|
|
|
|
|
def repair_authored_command(check_id: str, config: DoctorConfig) -> RepairResult:
|
|
command = REPAIR_COMMANDS[check_id]
|
|
exit_code, _ = run_repair_command(command, config)
|
|
message = "Repair completed. A fresh health check will verify recovery." if exit_code == 0 \
|
|
else "The authored repair command could not be completed."
|
|
return RepairResult(check_id, True, exit_code, message)
|
|
|
|
|
|
def lexical_path(path: Path) -> Path:
|
|
"""Normalize dot segments without following any filesystem symlink."""
|
|
return Path(os.path.abspath(os.fspath(path)))
|
|
|
|
|
|
def lexical_link_target(destination: Path) -> Path:
|
|
target = Path(os.readlink(destination))
|
|
return lexical_path(target if target.is_absolute() else destination.parent / target)
|
|
|
|
|
|
def renameat2(source: Path, destination: Path, flags: int) -> None:
|
|
"""Call Linux renameat2 with fixed flags selected by authored code."""
|
|
libc = ctypes.CDLL(None, use_errno=True)
|
|
function = getattr(libc, "renameat2", None)
|
|
if function is None:
|
|
raise OSError(errno.ENOSYS, "renameat2 is unavailable")
|
|
function.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
|
|
function.restype = ctypes.c_int
|
|
result = function(
|
|
AT_FDCWD,
|
|
os.fsencode(source),
|
|
AT_FDCWD,
|
|
os.fsencode(destination),
|
|
flags,
|
|
)
|
|
if result != 0:
|
|
error = ctypes.get_errno()
|
|
raise OSError(error, os.strerror(error), destination)
|
|
|
|
|
|
def rename_exchange(source: Path, destination: Path) -> None:
|
|
renameat2(source, destination, RENAME_EXCHANGE)
|
|
|
|
|
|
def rename_noreplace(source: Path, destination: Path) -> None:
|
|
renameat2(source, destination, RENAME_NOREPLACE)
|
|
|
|
|
|
def create_symlink_candidate(destination: Path, source: Path) -> Path:
|
|
"""Create one unpredictable authored sibling candidate symlink."""
|
|
for _ in range(32):
|
|
candidate = destination.with_name(
|
|
f".panama-link-{destination.name}-{os.getpid()}-{secrets.token_hex(8)}"
|
|
)
|
|
try:
|
|
os.symlink(source, candidate, target_is_directory=True)
|
|
return candidate
|
|
except FileExistsError:
|
|
continue
|
|
raise OSError("Could not allocate an authored temporary link")
|
|
|
|
|
|
def cleanup_candidate(candidate: Path) -> None:
|
|
try:
|
|
candidate.unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def install_absent_symlink(destination: Path, source: Path) -> Literal["repaired", "blocked", "failed"]:
|
|
candidate = create_symlink_candidate(destination, source)
|
|
try:
|
|
try:
|
|
rename_noreplace(candidate, destination)
|
|
except FileExistsError:
|
|
return "blocked"
|
|
except OSError:
|
|
return "failed"
|
|
return "repaired"
|
|
finally:
|
|
cleanup_candidate(candidate)
|
|
|
|
|
|
def exchange_owned_symlink(
|
|
destination: Path,
|
|
source: Path,
|
|
authored_sources: frozenset[Path],
|
|
) -> Literal["repaired", "blocked", "failed"]:
|
|
"""Exchange first, then validate the exact object removed from destination."""
|
|
candidate = create_symlink_candidate(destination, source)
|
|
exchanged = False
|
|
rolled_back = False
|
|
try:
|
|
try:
|
|
rename_exchange(candidate, destination)
|
|
exchanged = True
|
|
except OSError:
|
|
return "failed"
|
|
|
|
try:
|
|
old_is_authored = candidate.is_symlink() \
|
|
and lexical_link_target(candidate) in authored_sources
|
|
except OSError:
|
|
old_is_authored = False
|
|
if old_is_authored:
|
|
cleanup_candidate(candidate)
|
|
return "repaired"
|
|
|
|
try:
|
|
rename_exchange(candidate, destination)
|
|
rolled_back = True
|
|
except OSError:
|
|
# The displaced object remains at the unpredictable candidate path;
|
|
# never unlink it when rollback could not restore ownership.
|
|
return "failed"
|
|
|
|
try:
|
|
restored_candidate_is_ours = candidate.is_symlink() \
|
|
and lexical_link_target(candidate) == source
|
|
except OSError:
|
|
restored_candidate_is_ours = False
|
|
if not restored_candidate_is_ours:
|
|
return "failed"
|
|
cleanup_candidate(candidate)
|
|
return "blocked"
|
|
finally:
|
|
if not exchanged or rolled_back:
|
|
try:
|
|
if candidate.is_symlink() and lexical_link_target(candidate) == source:
|
|
cleanup_candidate(candidate)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def repair_runtime_links(config: DoctorConfig) -> RepairResult:
|
|
root = lexical_path(config.root)
|
|
sources = [(name, lexical_path(config.root / relative_source)) for name, relative_source in RUNTIME_LINK_TARGETS]
|
|
if any(not source.is_dir() for _, source in sources):
|
|
return RepairResult("panama.runtime-links", True, 1, "Tracked Panama link destinations are unavailable.")
|
|
if any(not source.is_relative_to(root) for _, source in sources):
|
|
return RepairResult("panama.runtime-links", True, 1, "Tracked Panama link destinations are invalid.")
|
|
authored_sources = frozenset(source for _, source in sources)
|
|
|
|
try:
|
|
config.config_home.mkdir(parents=True, exist_ok=True)
|
|
except OSError:
|
|
return RepairResult("panama.runtime-links", True, 1, "Panama runtime links could not be accessed.")
|
|
|
|
blocked = False
|
|
failed = False
|
|
for name, source in sources:
|
|
destination = config.config_home / name
|
|
try:
|
|
if destination.is_symlink():
|
|
current_target = lexical_link_target(destination)
|
|
if current_target == source:
|
|
continue
|
|
if current_target not in authored_sources:
|
|
blocked = True
|
|
continue
|
|
outcome = exchange_owned_symlink(destination, source, authored_sources)
|
|
blocked = blocked or outcome == "blocked"
|
|
failed = failed or outcome == "failed"
|
|
elif destination.exists():
|
|
# A regular file or directory is user-owned unless proven
|
|
# otherwise. Report it, but never replace it.
|
|
blocked = True
|
|
else:
|
|
outcome = install_absent_symlink(destination, source)
|
|
blocked = blocked or outcome == "blocked"
|
|
failed = failed or outcome == "failed"
|
|
except OSError:
|
|
failed = True
|
|
|
|
if failed:
|
|
return RepairResult("panama.runtime-links", True, 1, "One or more Panama runtime links could not be recreated.")
|
|
if blocked:
|
|
return RepairResult("panama.runtime-links", True, 1, "A user-owned runtime path is blocking a Panama link.")
|
|
return RepairResult("panama.runtime-links", True, 0, "Panama runtime links were recreated. A fresh health check will verify them.")
|
|
|
|
|
|
def repair_vicinae_commands(config: DoctorConfig) -> RepairResult:
|
|
helper = config.root / "setup/scripts/link-vicinae-scripts"
|
|
if not helper.is_file():
|
|
return RepairResult("panama.vicinae-commands", True, 127, "The authored Vicinae link helper is unavailable.")
|
|
exit_code, _ = run_repair_command((str(helper),), config, config.root)
|
|
message = "Panama commands were relinked. A fresh health check will verify them." if exit_code == 0 \
|
|
else "Panama commands could not be relinked."
|
|
return RepairResult("panama.vicinae-commands", True, exit_code, message)
|
|
|
|
|
|
def parse_caffeine_rows(output: str, uid: str) -> list[InhibitorRow] | None:
|
|
inhibitor_rows: list[InhibitorRow] = []
|
|
for line in output.splitlines():
|
|
parts = line.split()
|
|
if len(parts) < 2 or parts[0] != "Panama" or parts[1] != uid:
|
|
continue
|
|
if len(parts) != 8:
|
|
if "Caffeine" in parts:
|
|
return None
|
|
continue
|
|
if parts[6] != "Caffeine" or parts[7] != "block":
|
|
continue
|
|
if not parts[3].isdecimal():
|
|
return None
|
|
inhibitor_rows.append(tuple(parts))
|
|
return inhibitor_rows
|
|
|
|
|
|
def close_pidfds(pidfds: list[int]) -> None:
|
|
for pidfd in pidfds:
|
|
try:
|
|
os.close(pidfd)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def signal_caffeine_pidfds(
|
|
pidfds: list[int],
|
|
sender: Callable[..., None] | None = None,
|
|
) -> Literal["released", "preflight-failed", "incomplete"]:
|
|
send = sender or signal.pidfd_send_signal
|
|
for pidfd in pidfds:
|
|
try:
|
|
send(pidfd, 0, None, 0)
|
|
except (OSError, ValueError):
|
|
return "preflight-failed"
|
|
|
|
incomplete = False
|
|
for pidfd in pidfds:
|
|
try:
|
|
send(pidfd, signal.SIGTERM, None, 0)
|
|
except ProcessLookupError:
|
|
continue
|
|
except OSError as error:
|
|
if error.errno != errno.ESRCH:
|
|
incomplete = True
|
|
except ValueError:
|
|
incomplete = True
|
|
return "incomplete" if incomplete else "released"
|
|
|
|
|
|
def repair_caffeine(config: DoctorConfig) -> RepairResult:
|
|
list_command = ("systemd-inhibit", "--list", "--no-pager", "--no-legend")
|
|
list_exit, output = run_repair_command(list_command, config)
|
|
if list_exit != 0:
|
|
return RepairResult("panama.caffeine", True, list_exit, "Caffeine inhibitors could not be inspected.")
|
|
|
|
uid = str(os.getuid())
|
|
inhibitor_rows = parse_caffeine_rows(output, uid)
|
|
if inhibitor_rows is None:
|
|
return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata was invalid; nothing was released.")
|
|
inhibitor_pids = list(dict.fromkeys(int(row[3]) for row in inhibitor_rows))
|
|
|
|
if len(inhibitor_pids) <= 1:
|
|
return RepairResult("panama.caffeine", True, 0, "No duplicate Panama Caffeine inhibitors needed release.")
|
|
|
|
duplicates = inhibitor_pids[1:]
|
|
if not hasattr(os, "pidfd_open") or not hasattr(signal, "pidfd_send_signal"):
|
|
return RepairResult("panama.caffeine", True, 1, "Safe Caffeine inhibitor release is unavailable on this system.")
|
|
|
|
pidfds: list[int] = []
|
|
try:
|
|
try:
|
|
pidfds = [os.pidfd_open(pid, 0) for pid in duplicates]
|
|
except (OSError, ValueError):
|
|
return RepairResult("panama.caffeine", True, 1, "A duplicate inhibitor changed before it could be safely released.")
|
|
|
|
second_exit, second_output = run_repair_command(list_command, config)
|
|
if second_exit != 0:
|
|
return RepairResult("panama.caffeine", True, second_exit, "Caffeine inhibitors could not be revalidated; nothing was released.")
|
|
second_rows = parse_caffeine_rows(second_output, uid)
|
|
if second_rows is None or second_rows != inhibitor_rows:
|
|
return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata changed; nothing was released.")
|
|
|
|
signal_outcome = signal_caffeine_pidfds(pidfds)
|
|
if signal_outcome == "preflight-failed":
|
|
return RepairResult("panama.caffeine", True, 1, "A duplicate inhibitor changed before it could be safely released.")
|
|
if signal_outcome == "incomplete":
|
|
return RepairResult("panama.caffeine", True, 1, "One or more duplicate inhibitors could not be released.")
|
|
finally:
|
|
close_pidfds(pidfds)
|
|
return RepairResult("panama.caffeine", True, 0, "Duplicate Panama Caffeine inhibitors were released. A fresh health check will verify recovery.")
|
|
|
|
|
|
def repair(check_id: str, config: DoctorConfig) -> RepairResult:
|
|
if check_id in REPAIR_COMMANDS:
|
|
return repair_authored_command(check_id, config)
|
|
if check_id == "panama.runtime-links":
|
|
return repair_runtime_links(config)
|
|
if check_id == "panama.vicinae-commands":
|
|
return repair_vicinae_commands(config)
|
|
if check_id == "panama.caffeine":
|
|
return repair_caffeine(config)
|
|
return RepairResult(check_id, False, 2, "This health check has no authored repair.")
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
parser = argparse.ArgumentParser(description="Panama system diagnostics and bounded repairs")
|
|
output = parser.add_mutually_exclusive_group()
|
|
output.add_argument("--json", action="store_true")
|
|
output.add_argument("--summary", action="store_true")
|
|
parser.add_argument("--repair", metavar="CHECK_ID")
|
|
args = parser.parse_args(argv)
|
|
|
|
if args.repair is not None:
|
|
if args.repair not in REPAIR_IDS or args.summary:
|
|
result = RepairResult(args.repair, False, 2, "This health check has no authored repair.")
|
|
else:
|
|
try:
|
|
result = repair(args.repair, config_from_environment())
|
|
except Exception:
|
|
result = RepairResult(args.repair, True, 1, "The authored repair could not be completed.")
|
|
print(json.dumps(result.as_json(), separators=(",", ":"), sort_keys=False))
|
|
return result.exit_code
|
|
|
|
result = snapshot(config_from_environment())
|
|
if args.summary:
|
|
summary = result["summary"]
|
|
assert isinstance(summary, dict)
|
|
print(f"Panama system health: {summary['status']} ({summary['healthy']} ok, {summary['warnings']} warnings, {summary['errors']} errors, {summary['unconfigured']} unconfigured)")
|
|
else:
|
|
print(json.dumps(result, separators=(",", ":"), sort_keys=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|