ops-bridge/src/bridge/state.py

"""State file management for OpsBridge."""
from __future__ import annotations

import os
from pathlib import Path
from typing import Optional

from bridge.models import BridgeState


def _default_state_dir() -> Path:
    return Path.home() / ".local" / "state" / "bridge"


class StateManager:
    def __init__(self, state_dir: Optional[Path] = None):
        self._dir = Path(state_dir) if state_dir else _default_state_dir()

    def _ensure_dir(self) -> None:
        self._dir.mkdir(parents=True, exist_ok=True)

    def _state_path(self, name: str) -> Path:
        return self._dir / f"{name}.state"

    def _pid_path(self, name: str) -> Path:
        return self._dir / f"{name}.pid"

    def read_state(self, name: str) -> BridgeState:
        path = self._state_path(name)
        if not path.exists():
            return BridgeState.STOPPED
        text = path.read_text().strip()
        try:
            return BridgeState(text)
        except ValueError:
            return BridgeState.STOPPED

    def write_state(self, name: str, state: BridgeState) -> None:
        self._ensure_dir()
        self._state_path(name).write_text(state.value)

    def read_pid(self, name: str) -> Optional[int]:
        path = self._pid_path(name)
        if not path.exists():
            return None
        try:
            pid = int(path.read_text().strip())
        except (ValueError, OSError):
            return None
        if _pid_alive(pid):
            return pid
        return None

    def read_raw_pid(self, name: str) -> Optional[int]:
        """Read PID from file without liveness check. Returns None if file absent/invalid."""
        path = self._pid_path(name)
        if not path.exists():
            return None
        try:
            return int(path.read_text().strip())
        except (ValueError, OSError):
            return None

    def write_pid(self, name: str, pid: int) -> None:
        self._ensure_dir()
        self._pid_path(name).write_text(str(pid))

    def clear_pid(self, name: str) -> None:
        path = self._pid_path(name)
        if path.exists():
            path.unlink()

    def is_running(self, name: str) -> bool:
        return self.read_pid(name) is not None


def _pid_alive(pid: int) -> bool:
    """Return True if the process with given PID exists."""
    try:
        os.kill(pid, 0)
        return True
    except (ProcessLookupError, PermissionError):
        return False