#!/usr/bin/env python3 """Redacted diagnostics and bounded repairs for Panama-owned functionality.""" from __future__ import annotations import argparse import json import os import re 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"] @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 = ("quickshell", "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" if source.is_dir() and installed.is_symlink() and installed.exists(): 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: counts: list[int] = [] 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.") uid = str(os.getuid()) inhibitors = 0 malformed = False for line in result.stdout.splitlines(): parts = line.split() relevant = len(parts) >= 2 and parts[0] == "Panama" and parts[1] == uid and "Caffeine" in parts if not relevant: continue if len(parts) >= 8 and parts[3].isdecimal() and parts[-2:] == ["Caffeine", "block"]: inhibitors += 1 else: malformed = True if malformed: return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe returned an invalid result.") 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 repair_runtime_links(config: DoctorConfig) -> RepairResult: sources = [(name, 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.") 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(): if destination.resolve(strict=False) == source.resolve(strict=True): continue destination.unlink() destination.symlink_to(source, target_is_directory=True) elif destination.exists(): # A regular file or directory is user-owned unless proven # otherwise. Report it, but never replace it. blocked = True else: destination.symlink_to(source, target_is_directory=True) 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 file or directory is blocking a Panama runtime 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 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_pids: list[str] = [] 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: return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata was invalid; nothing was released.") if parts[6] != "Caffeine" or parts[7] != "block": continue if not parts[3].isdecimal(): return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata was invalid; nothing was released.") inhibitor_pids.append(parts[3]) if len(inhibitor_pids) <= 1: return RepairResult("panama.caffeine", True, 0, "No duplicate Panama Caffeine inhibitors needed release.") for pid in inhibitor_pids[1:]: exit_code, _ = run_repair_command(("kill", "--", pid), config) if exit_code != 0: return RepairResult("panama.caffeine", True, exit_code, "A duplicate Panama Caffeine inhibitor could not be released.") 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:]))