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 <noreply@anthropic.com>
This commit is contained in:
curo1305
@@ -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"""<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
|
||||
"http://www.apple.com/DTDs/PropertyList-1.0.dtd">
|
||||
<plist version="1.0">
|
||||
<dict>
|
||||
<key>Label</key>
|
||||
<string>com.pyra.daemon</string>
|
||||
<key>ProgramArguments</key>
|
||||
<array>
|
||||
<string>{exe}</string>
|
||||
<string>daemon</string>
|
||||
<string>run</string>
|
||||
</array>
|
||||
<key>RunAtLoad</key>
|
||||
<true/>
|
||||
<key>KeepAlive</key>
|
||||
<true/>
|
||||
<key>StandardOutPath</key>
|
||||
<string>{log}</string>
|
||||
<key>StandardErrorPath</key>
|
||||
<string>{log}</string>
|
||||
<key>ProcessType</key>
|
||||
<string>Background</string>
|
||||
</dict>
|
||||
</plist>
|
||||
"""
|
||||
|
||||
|
||||
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"""<?xml version="1.0" encoding="UTF-16"?>
|
||||
<Task version="1.2" xmlns="http://schemas.microsoft.com/windows/2004/02/mit/task">
|
||||
<RegistrationInfo>
|
||||
<Description>Pyra Personal AI Assistant Daemon</Description>
|
||||
</RegistrationInfo>
|
||||
<Triggers>
|
||||
<LogonTrigger>
|
||||
<Enabled>true</Enabled>
|
||||
</LogonTrigger>
|
||||
</Triggers>
|
||||
<Settings>
|
||||
<MultipleInstancesPolicy>IgnoreNew</MultipleInstancesPolicy>
|
||||
<DisallowStartIfOnBatteries>false</DisallowStartIfOnBatteries>
|
||||
<StopIfGoingOnBatteries>false</StopIfGoingOnBatteries>
|
||||
<ExecutionTimeLimit>PT0S</ExecutionTimeLimit>
|
||||
<RestartOnFailure>
|
||||
<Interval>PT1M</Interval>
|
||||
<Count>999</Count>
|
||||
</RestartOnFailure>
|
||||
</Settings>
|
||||
<Actions Context="Author">
|
||||
<Exec>
|
||||
<Command>{exe}</Command>
|
||||
<Arguments>daemon run</Arguments>
|
||||
</Exec>
|
||||
</Actions>
|
||||
</Task>
|
||||
"""
|
||||
|
||||
|
||||
# ── 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
|
||||
)
|
||||
@@ -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 "<string>daemon</string>" in xml
|
||||
assert "<string>run</string>" in xml
|
||||
assert "com.pyra.daemon" in xml
|
||||
assert "<true/>" 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("<plist/>")
|
||||
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()
|
||||
Reference in New Issue
Block a user