Merge current Panama main

config/dot/quickshell/scripts/panama-doctor

@@ -5,9 +5,13 @@
 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
@@ -21,6 +25,11 @@ 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)
@@ -409,8 +418,13 @@ def check_runtime_links(config: DoctorConfig) -> Check:
 
 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():
+    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"))
 
@@ -445,20 +459,10 @@ 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:
+    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:
@@ -535,10 +539,138 @@ def repair_authored_command(check_id: str, config: DoctorConfig) -> RepairResult
     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:
-    sources = [(name, config.root / relative_source) for name, relative_source in RUNTIME_LINK_TARGETS]
+    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)
@@ -551,23 +683,30 @@ def repair_runtime_links(config: DoctorConfig) -> RepairResult:
         destination = config.config_home / name
         try:
             if destination.is_symlink():
-                if destination.resolve(strict=False) == source.resolve(strict=True):
+                current_target = lexical_link_target(destination)
+                if current_target == source:
                     continue
-                destination.unlink()
-                destination.symlink_to(source, target_is_directory=True)
+                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:
-                destination.symlink_to(source, target_is_directory=True)
+                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 file or directory is blocking a Panama runtime link.")
+        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.")
 
 
@@ -581,6 +720,57 @@ def repair_vicinae_commands(config: DoctorConfig) -> RepairResult:
     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)
@@ -588,26 +778,39 @@ def repair_caffeine(config: DoctorConfig) -> RepairResult:
         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])
+    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.")
 
-    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.")
+    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.")
 
 

config/dot/quickshell/scripts/panama-kdeconnect

@@ -25,6 +25,9 @@ PLUGIN_ACTIONS = {
     "kdeconnect_share": "share",
 }
 DEVICE_OBJECT_PREFIX = "/modules/kdeconnect/devices"
+DEVICE_OBJECT_LINE = re.compile(
+    rf"(?P<path>{re.escape(DEVICE_OBJECT_PREFIX)}/(?P<id>[A-Fa-f0-9]{{32,64}}))$"
+)
 Runner = Callable[..., subprocess.CompletedProcess[str]]
 
 
@@ -107,6 +110,13 @@ def parse_string_property(output: str) -> str:
     return parts[1] if len(parts) == 2 and parts[0] == "s" else ""
 
 
+def parse_bool_property(output: str) -> bool | None:
+    parts = output.split()
+    if len(parts) != 2 or parts[0] != "b" or parts[1] not in {"true", "false"}:
+        return None
+    return parts[1] == "true"
+
+
 def run_command(
     command: list[str],
     *,
@@ -179,6 +189,82 @@ def reported_type(device_id: str, runner: Runner = subprocess.run) -> str:
     return parse_string_property(result.stdout) if result.returncode == 0 else ""
 
 
+def device_property(
+    device_id: str,
+    member: str,
+    runner: Runner = subprocess.run,
+) -> subprocess.CompletedProcess[str]:
+    return run_command(
+        [
+            "busctl",
+            "--user",
+            "get-property",
+            "org.kde.kdeconnect",
+            device_object(device_id),
+            "org.kde.kdeconnect.device",
+            member,
+        ],
+        runner=runner,
+    )
+
+
+def dbus_device_ids(runner: Runner = subprocess.run) -> list[str]:
+    try:
+        result = run_command(
+            ["busctl", "--user", "tree", "org.kde.kdeconnect"],
+            runner=runner,
+        )
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        return []
+    if result.returncode != 0:
+        return []
+    return [
+        match.group("id")
+        for line in result.stdout.splitlines()
+        if (match := DEVICE_OBJECT_LINE.search(line.strip())) is not None
+    ]
+
+
+def dbus_devices(runner: Runner = subprocess.run) -> list[dict[str, object]]:
+    devices: list[dict[str, object]] = []
+    for device_id in dbus_device_ids(runner):
+        try:
+            name_result = device_property(device_id, "name", runner)
+            type_result = device_property(device_id, "type", runner)
+            paired_result = device_property(device_id, "isPaired", runner)
+            reachable_result = device_property(device_id, "isReachable", runner)
+        except (FileNotFoundError, subprocess.TimeoutExpired):
+            continue
+        name = parse_string_property(name_result.stdout) if name_result.returncode == 0 else ""
+        device_type = parse_string_property(type_result.stdout) if type_result.returncode == 0 else ""
+        paired = parse_bool_property(paired_result.stdout) if paired_result.returncode == 0 else None
+        reachable = parse_bool_property(reachable_result.stdout) if reachable_result.returncode == 0 else None
+        if not name or paired is not True or reachable is None:
+            continue
+        try:
+            plugins = device_plugins(device_id, runner)
+        except (FileNotFoundError, subprocess.TimeoutExpired):
+            plugins = []
+        actions = sorted(
+            {
+                action
+                for plugin, action in PLUGIN_ACTIONS.items()
+                if plugin in plugins
+            }
+        )
+        devices.append(
+            {
+                "id": device_id,
+                "name": name,
+                "type": device_type or inferred_type(name),
+                "paired": paired,
+                "reachable": reachable,
+                "actions": actions,
+            }
+        )
+    return devices
+
+
 def collect_status(runner: Runner = subprocess.run) -> dict[str, object]:
     try:
         listing = run_command(
@@ -200,15 +286,24 @@ def collect_status(runner: Runner = subprocess.run) -> dict[str, object]:
             continue
         device_id = match.group("id")
         try:
-            device = normalize_device_line(
-                line,
-                device_plugins(device_id, runner),
-                reported_type(device_id, runner),
-            )
+            plugins = device_plugins(device_id, runner)
+            device_type = reported_type(device_id, runner)
+        except (FileNotFoundError, subprocess.TimeoutExpired):
+            plugins = []
+            device_type = ""
+        try:
+            device = normalize_device_line(line, plugins, device_type)
         except ValueError:
             continue
         devices.append(device)
 
+    known_ids = {str(device["id"]) for device in devices}
+    devices.extend(
+        device
+        for device in dbus_devices(runner)
+        if str(device["id"]) not in known_ids
+    )
+
     devices.sort(
         key=lambda device: (
             not bool(device["reachable"]),