feat(daemon): add PID file management module

Atomic write-then-rename, stale-PID detection via os.kill on POSIX and
ctypes.OpenProcess on Windows, context manager for cleanup on exit.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
curo1305
2026-05-19 15:22:04 +02:00
parent 0e052c4992
commit eaed52006f
2 changed files with 197 additions and 0 deletions
+94
View File
@@ -0,0 +1,94 @@
"""PID file management for the Pyra daemon."""
from __future__ import annotations
import os
import sys
from pathlib import Path
class PidFileError(OSError):
"""Raised when a PID file operation fails due to a live conflicting process."""
class PidFile:
def __init__(self, path: Path) -> None:
self._path = path
def write(self) -> None:
"""Write the current PID atomically.
Raises PidFileError if a non-stale PID file already exists.
"""
existing = self.read()
if existing is not None and not self.is_stale():
raise PidFileError(
f"Daemon already running with PID {existing} "
f"(PID file: {self._path})"
)
tmp = self._path.with_suffix(".pid.tmp")
tmp.write_text(str(os.getpid()))
tmp.replace(self._path)
def read(self) -> int | None:
"""Return the PID from the file, or None if the file is absent or unreadable."""
try:
return int(self._path.read_text().strip())
except (FileNotFoundError, ValueError):
return None
def is_stale(self) -> bool:
"""True when the PID file exists but the process no longer runs."""
pid = self.read()
if pid is None:
return False
return not _process_is_alive(pid)
def remove(self) -> None:
"""Delete the PID file, ignoring FileNotFoundError."""
try:
self._path.unlink()
except FileNotFoundError:
pass
def __enter__(self) -> "PidFile":
self.write()
return self
def __exit__(self, *_: object) -> None:
self.remove()
def resolve_pid_path(cfg_path: str) -> Path:
"""Expand ~ and return an absolute Path."""
return Path(cfg_path).expanduser().resolve()
# ── Platform-specific process liveness check ─────────────────────────────────
def _process_is_alive(pid: int) -> bool:
if sys.platform == "win32":
return _win_process_is_alive(pid)
return _posix_process_is_alive(pid)
def _posix_process_is_alive(pid: int) -> bool:
try:
os.kill(pid, 0)
return True
except ProcessLookupError:
return False
except PermissionError:
# Process exists but is owned by another user — still alive.
return True
def _win_process_is_alive(pid: int) -> bool:
import ctypes
SYNCHRONIZE = 0x00100000
handle = ctypes.windll.kernel32.OpenProcess(SYNCHRONIZE, False, pid) # type: ignore[attr-defined]
if handle == 0:
return False
ctypes.windll.kernel32.CloseHandle(handle) # type: ignore[attr-defined]
return True