solpbc.org
linux observer
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Desktop session environment checks and recovery.
5
6Extracted from solstone's observe/linux/observer.py (lines 598-666).
7
8_recover_session_env() is kept as fallback for manual CLI launch.
9For systemd service launch, PassEnvironment= in the unit file is
10the primary mechanism.
11"""
12
13import logging
14import os
15import shutil
16import subprocess
17
18logger = logging.getLogger(__name__)
19
20# Exit codes
21EXIT_TEMPFAIL = 75 # EX_TEMPFAIL: session not ready, retry later
22
23
24def _recover_session_env() -> None:
25 """Try to recover desktop session env vars from the systemd user manager.
26
27 On GNOME Wayland, gnome-shell pushes DISPLAY, WAYLAND_DISPLAY, and
28 DBUS_SESSION_BUS_ADDRESS into the systemd user environment on startup.
29 When the observer is launched from a non-desktop shell, these vars may be missing
30 from the inherited environment — but systemctl --user show-environment
31 has them.
32 """
33 needed = {"DISPLAY", "WAYLAND_DISPLAY", "DBUS_SESSION_BUS_ADDRESS"}
34 missing = {v for v in needed if not os.environ.get(v)}
35 if not missing:
36 return
37
38 # Ensure XDG_RUNTIME_DIR is set (required for systemctl --user to connect)
39 if not os.environ.get("XDG_RUNTIME_DIR"):
40 os.environ["XDG_RUNTIME_DIR"] = f"/run/user/{os.getuid()}"
41
42 try:
43 result = subprocess.run(
44 ["systemctl", "--user", "show-environment"],
45 capture_output=True,
46 text=True,
47 timeout=5,
48 )
49 if result.returncode != 0:
50 return
51 except (FileNotFoundError, subprocess.TimeoutExpired):
52 return
53
54 recovered = []
55 for line in result.stdout.splitlines():
56 key, _, value = line.partition("=")
57 if key in missing and value:
58 os.environ[key] = value
59 recovered.append(f"{key}={value}")
60
61 if recovered:
62 logger.info("Recovered session env from systemd: %s", ", ".join(recovered))
63
64
65def check_session_ready() -> str | None:
66 """Check if the desktop session is ready for observation.
67
68 Returns None if ready, or a description of what's missing.
69 """
70 # Try to recover missing session vars from systemd user manager
71 _recover_session_env()
72
73 # Display server
74 if not os.environ.get("DISPLAY") and not os.environ.get("WAYLAND_DISPLAY"):
75 return "no display server (DISPLAY/WAYLAND_DISPLAY not set)"
76
77 # DBus session bus
78 if not os.environ.get("DBUS_SESSION_BUS_ADDRESS"):
79 return "no DBus session bus (DBUS_SESSION_BUS_ADDRESS not set)"
80
81 # PulseAudio / PipeWire audio
82 pactl = shutil.which("pactl")
83 if pactl:
84 try:
85 subprocess.run(
86 [pactl, "info"],
87 capture_output=True,
88 timeout=5,
89 ).check_returncode()
90 except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
91 return "audio server not responding (pactl info failed)"
92 return None