From 513871ef96d0cf06bfc36b560bfb1a6c774faeec Mon Sep 17 00:00:00 2001 From: curo1305 Date: Tue, 19 May 2026 15:23:11 +0200 Subject: [PATCH] feat(daemon): add OS service install/uninstall module Generates launchd plist (macOS), systemd user unit (Linux), and Task Scheduler XML (Windows). Auto-detects platform; finds pyra executable via shutil.which with venv-sibling fallback. Co-Authored-By: Claude Sonnet 4.6 --- src/pyra/daemon/service.py | 212 ++++++++++++++++++++++++++++++ tests/unit/test_daemon_service.py | 189 ++++++++++++++++++++++++++ 2 files changed, 401 insertions(+) create mode 100644 src/pyra/daemon/service.py create mode 100644 tests/unit/test_daemon_service.py diff --git a/src/pyra/daemon/service.py b/src/pyra/daemon/service.py new file mode 100644 index 0000000..6bb0f58 --- /dev/null +++ b/src/pyra/daemon/service.py @@ -0,0 +1,212 @@ +"""OS-specific service file generation and install/uninstall for the Pyra daemon.""" + +from __future__ import annotations + +import platform +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Literal + +from pyra.utils.paths import safe_chmod + + +def detect_platform() -> Literal["macos", "linux", "windows"]: + s = platform.system() + if s == "Darwin": + return "macos" + if s == "Linux": + return "linux" + if s == "Windows": + return "windows" + raise RuntimeError(f"Unsupported platform: {s}") + + +def find_pyra_executable() -> str: + """Return the full path to the active pyra executable. + + Tries, in order: + 1. shutil.which("pyra") — works when pyra is on PATH (activated venv) + 2. sys.executable's sibling "pyra" script — covers editable installs + 3. Fallback: sys.executable -m pyra + """ + found = shutil.which("pyra") + if found: + return found + + sibling = Path(sys.executable).parent / "pyra" + if sibling.exists(): + return str(sibling) + + return f"{sys.executable} -m pyra" + + +# ── Template generators ─────────────────────────────────────────────────────── + +def render_launchd_plist(exe: str, log_file: str, pid_file: str) -> str: + log = str(Path(log_file).expanduser()) + return f""" + + + + Label + com.pyra.daemon + ProgramArguments + + {exe} + daemon + run + + RunAtLoad + + KeepAlive + + StandardOutPath + {log} + StandardErrorPath + {log} + ProcessType + Background + + +""" + + +def render_systemd_unit(exe: str, log_file: str) -> str: + log = str(Path(log_file).expanduser()) + return f"""[Unit] +Description=Pyra Personal AI Assistant Daemon +After=default.target + +[Service] +Type=simple +ExecStart={exe} daemon run +Restart=on-failure +RestartSec=5s +StandardOutput=append:{log} +StandardError=append:{log} + +[Install] +WantedBy=default.target +""" + + +def render_schtasks_xml(exe: str) -> str: + return f""" + + + Pyra Personal AI Assistant Daemon + + + + true + + + + IgnoreNew + false + false + PT0S + + PT1M + 999 + + + + + {exe} + daemon run + + + +""" + + +# ── Install / uninstall ─────────────────────────────────────────────────────── + +def install_service() -> None: + """Generate and register the OS service for the current platform.""" + from pyra.config.manager import load_config + + cfg = load_config() + exe = find_pyra_executable() + plat = detect_platform() + + if plat == "macos": + _install_launchd(exe, cfg.daemon.log_file, cfg.daemon.pid_file) + elif plat == "linux": + _install_systemd(exe, cfg.daemon.log_file) + else: + _install_windows(exe) + + +def uninstall_service() -> None: + """Deregister the OS service for the current platform.""" + plat = detect_platform() + if plat == "macos": + _uninstall_launchd() + elif plat == "linux": + _uninstall_systemd() + else: + _uninstall_windows() + + +# ── macOS launchd ───────────────────────────────────────────────────────────── + +_PLIST_PATH = Path.home() / "Library" / "LaunchAgents" / "com.pyra.daemon.plist" + + +def _install_launchd(exe: str, log_file: str, pid_file: str) -> None: + _PLIST_PATH.parent.mkdir(parents=True, exist_ok=True) + _PLIST_PATH.write_text(render_launchd_plist(exe, log_file, pid_file)) + safe_chmod(_PLIST_PATH, 0o644) # launchd requires 644, not 600 + subprocess.run(["launchctl", "load", str(_PLIST_PATH)], check=True) + + +def _uninstall_launchd() -> None: + if _PLIST_PATH.exists(): + subprocess.run(["launchctl", "unload", str(_PLIST_PATH)], check=False) + _PLIST_PATH.unlink() + + +# ── Linux systemd ───────────────────────────────────────────────────────────── + +_SYSTEMD_UNIT = Path.home() / ".config" / "systemd" / "user" / "pyra.service" + + +def _install_systemd(exe: str, log_file: str) -> None: + _SYSTEMD_UNIT.parent.mkdir(parents=True, exist_ok=True) + _SYSTEMD_UNIT.write_text(render_systemd_unit(exe, log_file)) + subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) + subprocess.run(["systemctl", "--user", "enable", "pyra"], check=True) + + +def _uninstall_systemd() -> None: + subprocess.run( + ["systemctl", "--user", "disable", "--now", "pyra"], check=False + ) + if _SYSTEMD_UNIT.exists(): + _SYSTEMD_UNIT.unlink() + subprocess.run(["systemctl", "--user", "daemon-reload"], check=False) + + +# ── Windows Task Scheduler ──────────────────────────────────────────────────── + +def _install_windows(exe: str) -> None: + from pyra.utils.paths import pyra_home + + xml_path = pyra_home() / "daemon_task.xml" + # schtasks /Create /XML requires UTF-16 encoding + xml_path.write_text(render_schtasks_xml(exe), encoding="utf-16") + subprocess.run( + ["schtasks", "/Create", "/TN", "PyraAssistant", "/XML", str(xml_path), "/F"], + check=True, + ) + + +def _uninstall_windows() -> None: + subprocess.run( + ["schtasks", "/Delete", "/TN", "PyraAssistant", "/F"], check=False + ) diff --git a/tests/unit/test_daemon_service.py b/tests/unit/test_daemon_service.py new file mode 100644 index 0000000..0dc951a --- /dev/null +++ b/tests/unit/test_daemon_service.py @@ -0,0 +1,189 @@ +"""Unit tests for daemon service file generation and platform detection.""" + +from __future__ import annotations + +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from pyra.daemon.service import ( + detect_platform, + find_pyra_executable, + render_launchd_plist, + render_systemd_unit, + render_schtasks_xml, +) + + +# ── Template rendering ──────────────────────────────────────────────────────── + +def test_render_launchd_plist_contains_exe() -> None: + xml = render_launchd_plist("/usr/local/bin/pyra", "~/.pyra/daemon.log", "~/.pyra/daemon.pid") + assert "/usr/local/bin/pyra" in xml + assert "daemon" in xml + assert "run" in xml + assert "com.pyra.daemon" in xml + assert "" in xml # KeepAlive and RunAtLoad + + +def test_render_launchd_plist_expands_log_tilde() -> None: + xml = render_launchd_plist("/bin/pyra", "~/.pyra/daemon.log", "~/.pyra/daemon.pid") + assert "~" not in xml + + +def test_render_systemd_unit_contains_exe() -> None: + unit = render_systemd_unit("/usr/local/bin/pyra", "~/.pyra/daemon.log") + assert "ExecStart=/usr/local/bin/pyra daemon run" in unit + assert "Restart=on-failure" in unit + assert "Type=simple" in unit + assert "WantedBy=default.target" in unit + + +def test_render_systemd_unit_expands_log_tilde() -> None: + unit = render_systemd_unit("/bin/pyra", "~/.pyra/daemon.log") + assert "~" not in unit + + +def test_render_schtasks_xml_contains_exe() -> None: + xml = render_schtasks_xml("C:\\Users\\test\\pyra.exe") + assert "C:\\Users\\test\\pyra.exe" in xml + assert "LogonTrigger" in xml + assert "daemon run" in xml + assert "IgnoreNew" in xml + + +def test_render_schtasks_xml_no_time_limit() -> None: + xml = render_schtasks_xml("pyra.exe") + assert "PT0S" in xml # ExecutionTimeLimit=PT0S means unlimited + + +# ── Platform detection ──────────────────────────────────────────────────────── + +def test_detect_platform_returns_known_value() -> None: + result = detect_platform() + assert result in ("macos", "linux", "windows") + + +@pytest.mark.parametrize("system,expected", [ + ("Darwin", "macos"), + ("Linux", "linux"), + ("Windows", "windows"), +]) +def test_detect_platform_maps_correctly(system: str, expected: str) -> None: + with patch("platform.system", return_value=system): + assert detect_platform() == expected + + +def test_detect_platform_raises_on_unknown() -> None: + with patch("platform.system", return_value="FreeBSD"): + with pytest.raises(RuntimeError, match="Unsupported platform"): + detect_platform() + + +# ── Executable detection ────────────────────────────────────────────────────── + +def test_find_pyra_executable_returns_string() -> None: + result = find_pyra_executable() + assert isinstance(result, str) + assert len(result) > 0 + + +def test_find_pyra_executable_uses_which_when_available(tmp_path: Path) -> None: + fake_pyra = tmp_path / "pyra" + fake_pyra.touch() + with patch("shutil.which", return_value=str(fake_pyra)): + assert find_pyra_executable() == str(fake_pyra) + + +def test_find_pyra_executable_falls_back_to_sibling(tmp_path: Path) -> None: + fake_python = tmp_path / "python3" + fake_pyra = tmp_path / "pyra" + fake_pyra.touch() + with patch("shutil.which", return_value=None): + with patch("sys.executable", str(fake_python)): + assert find_pyra_executable() == str(fake_pyra) + + +def test_find_pyra_executable_falls_back_to_module(tmp_path: Path) -> None: + fake_python = tmp_path / "python3" + with patch("shutil.which", return_value=None): + with patch("sys.executable", str(fake_python)): + result = find_pyra_executable() + assert result == f"{fake_python} -m pyra" + + +# ── Install / uninstall (subprocess mocked) ─────────────────────────────────── + +@pytest.mark.skipif(sys.platform == "win32", reason="launchd install is macOS-only") +def test_install_launchd_writes_plist_and_calls_launchctl( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import pyra.daemon.service as svc + + plist_path = tmp_path / "Library" / "LaunchAgents" / "com.pyra.daemon.plist" + monkeypatch.setattr(svc, "_PLIST_PATH", plist_path) + + calls: list[list[str]] = [] + monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: calls.append(cmd)) + + svc._install_launchd("/usr/local/bin/pyra", "~/.pyra/daemon.log", "~/.pyra/daemon.pid") + + assert plist_path.exists() + assert "com.pyra.daemon" in plist_path.read_text() + assert any("launchctl" in c[0] for c in calls) + + +@pytest.mark.skipif(sys.platform == "win32", reason="systemd install is Linux-only") +def test_install_systemd_writes_unit_and_calls_systemctl( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import pyra.daemon.service as svc + + unit_path = tmp_path / ".config" / "systemd" / "user" / "pyra.service" + monkeypatch.setattr(svc, "_SYSTEMD_UNIT", unit_path) + + calls: list[list[str]] = [] + monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: calls.append(cmd)) + + svc._install_systemd("/usr/local/bin/pyra", "~/.pyra/daemon.log") + + assert unit_path.exists() + assert "ExecStart" in unit_path.read_text() + assert any("systemctl" in c[0] for c in calls) + + +@pytest.mark.skipif(sys.platform == "win32", reason="launchd uninstall is macOS-only") +def test_uninstall_launchd_removes_plist( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import pyra.daemon.service as svc + + plist_path = tmp_path / "Library" / "LaunchAgents" / "com.pyra.daemon.plist" + plist_path.parent.mkdir(parents=True) + plist_path.write_text("") + monkeypatch.setattr(svc, "_PLIST_PATH", plist_path) + monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: None) + + svc._uninstall_launchd() + + assert not plist_path.exists() + + +@pytest.mark.skipif(sys.platform == "win32", reason="systemd uninstall is Linux-only") +def test_uninstall_systemd_removes_unit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import pyra.daemon.service as svc + + unit_path = tmp_path / ".config" / "systemd" / "user" / "pyra.service" + unit_path.parent.mkdir(parents=True) + unit_path.write_text("[Service]") + monkeypatch.setattr(svc, "_SYSTEMD_UNIT", unit_path) + monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: None) + + svc._uninstall_systemd() + + assert not unit_path.exists()