orchestrata_ultimate.refactor.codemod.daemon

Reference › orchestrata_ultimate › refactor › codemod › daemon

Lifecycle management for the codemod daemon, which parses a workspace once, serves it over a Unix socket and watches for file changes. scripts.codemod uses CodemodDaemon for its start, stop and status commands; the process it launches runs _daemon_entry.

The daemon is started only through a systemd user unit template (ADR-0067: daemons execute deployed code only). start(repo_root) creates the state directory, takes a file lock so two starts cannot race, refuses if a daemon already runs for that repository, checks that the deployed launcher bin/codemod-daemon exists under the tools prefix (raising RuntimeError and spawning nothing if it does not), starts the unit instance named by _unit_instance (derived with systemd-escape), waits for its main process id and records it in the PID file. stop stops the unit, with a legacy path that sends SIGTERM and then SIGKILL after a timeout; restart combines the two; status reports whether the daemon runs. _pid_is_codemod_daemon checks that a PID really belongs to a codemod daemon for the repo before it is acted on, and stale PID files are removed.

At a glance

Class diagram

Dependency diagram

API

Classes

CodemodDaemon

Decorators: traces(spec='110', story='US-110-013', reqs=['REQ-110-133', 'REQ-110-134', 'REQ-110-135', 'REQ-110-136', 'REQ-110-137']) · Line 46

Manages the codemod daemon lifecycle.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/daemon.py

"""SPEC-110 codemod engine — daemon lifecycle management.

Manages the long-running codemod daemon process that keeps the workspace
in memory between invocations. Handles start/stop/status/restart with
PID files, lock files, and graceful shutdown.

ADR-0067: the daemon executes DEPLOYED code only. It is never spawned from
this checkout's interpreter; ``start`` goes through the
``orchestrata-codemod-daemon@<escaped repo root>.service`` systemd template
whose ``ExecStart`` is ``<tools_prefix>/bin/codemod-daemon`` (written by
``deploy/install-tools.sh``), and refuses when that launcher is missing.
"""

from __future__ import annotations

import fcntl
import logging
import os
import signal
import subprocess
import time
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod._daemon_entry import read_deployed_commit

logger = logging.getLogger(__name__)

DAEMON_DIR = ".codemod_daemon"
SOCKET_NAME = "socket"
PID_NAME = "pid"
LOCK_NAME = "lock"
SHUTDOWN_TIMEOUT = 5.0
MAINPID_TIMEOUT = 10.0
START_SETTLE_SECONDS = 1.0
DEFAULT_TOOLS_PREFIX = Path("/opt/orchestrata-tools")
DEFAULT_UNIT_NAME = "orchestrata-codemod-daemon"
_DAEMON_ENTRY_MODULE = "orchestrata_ultimate.refactor.codemod._daemon_entry"


@traces(
    spec="110",
    story="US-110-013",
    reqs=["REQ-110-133", "REQ-110-134", "REQ-110-135", "REQ-110-136", "REQ-110-137"],
)
class CodemodDaemon:
    """Manages the codemod daemon lifecycle.

    The daemon parses the workspace once at startup, creates a Unix socket
    for JSON-RPC communication, and watches for file changes via inotify.
    """

    def __init__(
        self,
        *,
        tools_prefix: Path = DEFAULT_TOOLS_PREFIX,
        unit_name: str = DEFAULT_UNIT_NAME,
        shutdown_timeout: float = SHUTDOWN_TIMEOUT,
    ) -> None:
        """Initialize CodemodDaemon with no active process.

        Args:
            tools_prefix: The deployed-tools prefix (``deploy/install-tools.sh``
                writes ``bin/codemod-daemon`` and ``VERSION`` under it).
            unit_name: The systemd template name (instance ``unit@<path>``).
            shutdown_timeout: Seconds a legacy (non-unit) daemon gets after
                SIGTERM before SIGKILL.
        """
        self.shutdown_timeout = shutdown_timeout
        self._repo_root: Path | None = None
        self._daemon_dir: Path | None = None
        self._pid_file: Path | None = None
        self._lock_file: Path | None = None
        self._socket_path: Path | None = None
        self.tools_prefix = tools_prefix
        self.unit_name = unit_name
        self.launcher = tools_prefix / "bin" / "codemod-daemon"

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _unit_instance(self, repo_root: Path) -> str:
        """The systemd instance serving *repo_root*: ``unit@<escaped path>.service``."""
        try:
            escaped = subprocess.run(
                ["systemd-escape", "--path", str(repo_root)],
                capture_output=True,
                text=True,
                check=True,
            ).stdout.strip()
        except FileNotFoundError as e:
            raise RuntimeError(
                "systemd-escape is not installed; the codemod daemon is "
                f"started through systemd units only ({e})"
            ) from e
        except subprocess.CalledProcessError as e:
            raise RuntimeError(
                f"systemd-escape --path {repo_root} failed "
                f"(rc {e.returncode}): {e.stderr.strip()}"
            ) from e
        return f"{self.unit_name}@{escaped}.service"

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _unit_loaded(self, instance: str) -> bool:
        """True when systemd has *instance* loaded (its template is installed)."""
        return self._unit_property(instance, "LoadState") == "loaded"

    @staticmethod
    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _systemctl(*args: str) -> subprocess.CompletedProcess[str]:
        """Run ``systemctl --user <args>``; a missing binary reports rc 127."""
        try:
            return subprocess.run(
                ["systemctl", "--user", *args],
                capture_output=True,
                text=True,
                check=False,
            )
        except FileNotFoundError as e:
            return subprocess.CompletedProcess(
                ["systemctl", "--user", *args], 127, "", str(e)
            )

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _unit_property(self, instance: str, prop: str) -> str | None:
        """``systemctl --user show -p <prop> --value`` for *instance*, or None."""
        result = self._systemctl("show", "-p", prop, "--value", instance)
        value = result.stdout.strip()
        return value if result.returncode == 0 and value else None

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _unit_show(self, instance: str, *props: str) -> dict[str, str]:
        """One ``systemctl --user show -p <prop>...`` call parsed to key/value."""
        args: list[str] = []
        for prop in props:
            args.extend(["-p", prop])
        result = self._systemctl("show", *args, instance)
        parsed: dict[str, str] = {}
        if result.returncode == 0:
            for line in result.stdout.splitlines():
                key, sep, val = line.partition("=")
                if sep:
                    parsed[key] = val.strip()
        return parsed

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _wait_for_main_pid(self, instance: str) -> int:
        """Poll the unit's MainPID until it is non-zero (bounded).

        Each poll reads MainPID, ActiveState and SubState in one ``show`` call.
        For a ``Type=simple`` unit ``systemctl start`` returns once the main
        process is forked, so a zero MainPID with ActiveState ``failed`` or
        ``inactive``, or SubState ``auto-restart``, means the first process is
        already gone: that is reported at once. A unit in ``auto-restart``
        forks again after ``RestartSec``, but that is a second start, not this
        one.
        """
        deadline = time.monotonic() + MAINPID_TIMEOUT
        while True:
            props = self._unit_show(instance, "MainPID", "ActiveState", "SubState")
            main_pid = int(props.get("MainPID") or 0)
            if main_pid != 0:
                return main_pid
            state = props.get("ActiveState")
            if (
                state in {"failed", "inactive"}
                or props.get("SubState") == "auto-restart"
            ):
                raise RuntimeError(f"daemon {instance} failed before serving ({state})")
            if time.monotonic() >= deadline:
                raise RuntimeError(
                    f"daemon {instance} failed before serving "
                    f"(no MainPID within {MAINPID_TIMEOUT:.0f}s, ActiveState={state})"
                )
            time.sleep(0.1)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _settle_after_start(self, instance: str, pid: int) -> None:
        """Hold ``START_SETTLE_SECONDS`` and fail if the unit leaves ``active``.

        For a launcher that exits at once, the first ``show`` after
        ``systemctl start`` still reports the dead process's MainPID because
        systemd has not reaped it yet, so a non-zero MainPID alone does not
        prove the start. The settle has the same shape as the promoter's P6b
        verify hold: poll ActiveState and SubState every 0.1 s and raise the
        moment the unit is not ``active`` or is in ``auto-restart``.
        """
        deadline = time.monotonic() + START_SETTLE_SECONDS
        while True:
            props = self._unit_show(instance, "ActiveState", "SubState")
            state = props.get("ActiveState")
            if state != "active" or props.get("SubState") == "auto-restart":
                raise RuntimeError(f"daemon {instance} failed before serving ({state})")
            if time.monotonic() >= deadline:
                return
            time.sleep(0.1)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _refuse_if_running(self, instance: str, repo_root: Path) -> None:
        """Raise when a daemon already serves this repo; clear a stale PID file.

        The same unit-first rule ``status`` applies: when the systemd instance
        is loaded, its ``MainPID`` decides and the PID file is never consulted
        (it can be stale after a reboot or a unit restart, pointing at an
        unrelated live pid); a stale file under a loaded unit is removed. Only
        an unloaded unit falls back to the PID file, and then only a pid that
        really is a codemod daemon counts as running.

        Raises:
            RuntimeError: A daemon is already running for this repo.
        """
        if self._unit_loaded(instance):
            main_pid = self._unit_property(instance, "MainPID")
            pid = int(main_pid) if main_pid and int(main_pid) != 0 else None
            if pid is not None and _process_alive(pid):
                raise RuntimeError(
                    f"Codemod daemon already running (PID {pid}) "
                    f"for {self._repo_root}"
                )
            self._remove_stale_pid_file()
            return
        existing_pid = self._read_pid()
        if existing_pid is None:
            return
        if _pid_is_codemod_daemon(existing_pid, repo_root):
            raise RuntimeError(
                f"Codemod daemon already running (PID {existing_pid}) "
                f"for {self._repo_root}"
            )
        self._remove_stale_pid_file()

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _remove_stale_pid_file(self) -> None:
        """Drop a PID file that names no live daemon."""
        if self._pid_file is not None and self._pid_file.exists():
            logger.info("Removing stale PID file %s", self._pid_file)
            try:
                self._pid_file.unlink()
            except OSError as e:
                logger.warning("Error removing %s: %s", self._pid_file, e)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def start(self, repo_root: Path) -> None:
        """Start the codemod daemon through its systemd template instance.

        The deployed launcher parses the workspace, creates the Unix socket,
        and starts the inotify watcher; this method records the unit's
        MainPID in the PID file (the entry point does not write it).

        Args:
            repo_root: Repository root directory.

        Raises:
            RuntimeError: If a daemon is already running for this repo, the
                deployed launcher is missing (nothing is spawned then), or
                the unit fails to start.
        """
        self._repo_root = repo_root.resolve()
        self._daemon_dir = self._repo_root / DAEMON_DIR
        self._pid_file = self._daemon_dir / PID_NAME
        self._lock_file = self._daemon_dir / LOCK_NAME
        self._socket_path = self._daemon_dir / SOCKET_NAME

        self._daemon_dir.mkdir(parents=True, exist_ok=True)

        lock_fd = self._acquire_lock()
        if lock_fd is None:
            raise RuntimeError(
                f"Cannot acquire lock for {self._repo_root}; "
                "another daemon may be starting"
            )

        try:
            # The refusal runs UNDER the lock: two concurrent starts cannot
            # both pass the check, and nothing is spawned (no `systemctl
            # start`) before both the refusal and the launcher check pass.
            instance = self._unit_instance(self._repo_root)
            self._refuse_if_running(instance, self._repo_root)

            if not self.launcher.exists():
                raise RuntimeError(
                    f"codemod daemon is not deployed: {self.launcher} is "
                    "missing — run deploy/install-tools.sh (ADR-0067: daemons "
                    "execute deployed code only)"
                )

            started = self._systemctl("start", instance)
            if started.returncode != 0:
                raise RuntimeError(
                    f"systemctl --user start {instance} failed "
                    f"(rc {started.returncode}): {started.stderr.strip()}"
                )
            pid = self._wait_for_main_pid(instance)
            self._settle_after_start(instance, pid)
            self._write_pid(pid)
            logger.info(
                "Codemod daemon started (PID %d) as %s for %s",
                pid,
                instance,
                self._repo_root,
            )
        finally:
            self._release_lock(lock_fd)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def stop(self) -> None:
        """Stop the codemod daemon gracefully.

        When the systemd instance is loaded, ``systemctl --user stop`` is the
        only stop: after it succeeds the files are cleaned up and NO pid is
        signalled (the PID file can be stale after a unit restart, and a
        recycled pid is not ours to kill). Only when the unit is not loaded
        does a live PID-file pid — a daemon started before the unit-only
        launch — get SIGTERM, a wait of ``shutdown_timeout``, then SIGKILL.

        Raises:
            RuntimeError: ``systemctl stop`` failed for a loaded unit.
        """
        if self._pid_file is None:
            if self._repo_root is not None:
                self._daemon_dir = self._repo_root / DAEMON_DIR
                self._pid_file = self._daemon_dir / PID_NAME
                self._socket_path = self._daemon_dir / SOCKET_NAME
            else:
                logger.warning("No repo root set; cannot stop daemon")
                return

        if self._repo_root is not None:
            instance = self._unit_instance(self._repo_root)
            if self._unit_loaded(instance):
                stopped = self._systemctl("stop", instance)
                if stopped.returncode != 0:
                    raise RuntimeError(
                        f"systemctl --user stop {instance} failed "
                        f"(rc {stopped.returncode}): {stopped.stderr.strip()}"
                    )
                self._cleanup_files()
                logger.info("Codemod daemon unit %s stopped", instance)
                return

        pid = self._read_pid()
        if pid is None:
            logger.info("No daemon PID file found")
            return

        if not _process_alive(pid):
            logger.info("Daemon PID %d is not running; cleaning up", pid)
            self._cleanup_files()
            return

        if self._repo_root is None or not _pid_is_codemod_daemon(pid, self._repo_root):
            logger.warning(
                "PID %d in the PID file is not a codemod daemon; "
                "removing the stale file, signalling nothing",
                pid,
            )
            self._cleanup_files()
            return

        logger.info("Sending SIGTERM to daemon PID %d", pid)
        try:
            os.kill(pid, signal.SIGTERM)
        except ProcessLookupError:
            logger.info("Daemon PID %d already exited", pid)
            self._cleanup_files()
            return

        deadline = time.monotonic() + self.shutdown_timeout
        while time.monotonic() < deadline:
            if not _process_alive(pid):
                break
            time.sleep(0.2)
        else:
            logger.warning(
                "Daemon PID %d did not exit in %.1fs; sending SIGKILL",
                pid,
                self.shutdown_timeout,
            )
            try:
                os.kill(pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
            time.sleep(0.5)

        self._cleanup_files()
        logger.info("Codemod daemon stopped (PID %d)", pid)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def status(self) -> dict:
        """Return daemon status.

        Returns:
            Dict with pid, running, socket_path, the systemd ``unit`` instance
            and its ``active_state``, the deployment ``identity`` (code root,
            deployed commit, whether the launcher exists), and uptime if
            available.
        """
        identity = {
            "code_root": str(self.tools_prefix / "lib"),
            "deployed_commit": read_deployed_commit(self.tools_prefix / "VERSION"),
            "deployed": self.launcher.exists(),
        }
        if self._pid_file is None:
            if self._repo_root is not None:
                self._daemon_dir = self._repo_root / DAEMON_DIR
                self._pid_file = self._daemon_dir / PID_NAME
                self._socket_path = self._daemon_dir / SOCKET_NAME
            else:
                return {
                    "running": False,
                    "pid": None,
                    "socket_path": None,
                    "unit": None,
                    "active_state": None,
                    "identity": identity,
                }

        instance = (
            self._unit_instance(self._repo_root)
            if self._repo_root is not None
            else None
        )
        if instance is not None and self._unit_loaded(instance):
            # The unit is the source of truth: a systemd restart changes the
            # MainPID without touching the PID file.
            main_pid = self._unit_property(instance, "MainPID")
            pid = int(main_pid) if main_pid and int(main_pid) != 0 else None
        else:
            pid = self._read_pid()
        running = pid is not None and _process_alive(pid)
        socket_exists = self._socket_path is not None and self._socket_path.exists()

        result: dict = {
            "running": running,
            "pid": pid,
            "socket_path": str(self._socket_path) if self._socket_path else None,
            "socket_exists": socket_exists,
            "unit": instance,
            "active_state": (
                self._unit_property(instance, "ActiveState") if instance else None
            ),
            "identity": identity,
        }

        if running and pid is not None:
            try:
                import psutil

                proc = psutil.Process(pid)
                create_time = proc.create_time()
                result["uptime_seconds"] = time.time() - create_time
                result["memory_rss"] = proc.memory_info().rss
            except (ImportError, Exception) as e:
                logger.debug("Could not get process info: %s", e)

        return result

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def restart(self) -> None:
        """Restart the daemon (stop then start)."""
        if self._repo_root is None:
            raise RuntimeError("No repo root set; cannot restart daemon")
        repo_root = self._repo_root
        self.stop()
        self.start(repo_root)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-134"])
    def _read_pid(self) -> int | None:
        """Read PID from the PID file."""
        if self._pid_file is None or not self._pid_file.exists():
            return None
        try:
            text = self._pid_file.read_text(encoding="utf-8").strip()
            return int(text)
        except (ValueError, OSError) as e:
            logger.warning("Invalid PID file: %s", e)
            return None

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-134"])
    def _write_pid(self, pid: int) -> None:
        """Write PID to the PID file."""
        if self._pid_file is not None:
            self._pid_file.write_text(str(pid), encoding="utf-8")

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-134"])
    def _acquire_lock(self) -> int | None:
        """Acquire the lock file. Returns file descriptor or None on failure."""
        if self._lock_file is None:
            return None
        try:
            fd = os.open(str(self._lock_file), os.O_WRONLY | os.O_CREAT, 0o644)
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            return fd
        except OSError as e:
            logger.debug("Lock acquisition failed: %s", e)
            return None

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-134"])
    def _release_lock(self, fd: int) -> None:
        """Release the lock file."""
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
            os.close(fd)
        except OSError as e:
            logger.warning("Error releasing lock: %s", e)

    @traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
    def _cleanup_files(self) -> None:
        """Remove PID and socket files."""
        for path in (self._pid_file, self._socket_path, self._lock_file):
            if path is not None and path.exists():
                try:
                    path.unlink()
                except OSError as e:
                    logger.warning("Error removing %s: %s", path, e)


@traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
def _pid_is_codemod_daemon(pid: int, repo_root: Path) -> bool:
    """True when *pid*'s argv runs the codemod daemon entry module for *repo_root*.

    The deployed launcher execs ``python -m <_DAEMON_ENTRY_MODULE> <repo_root>
    <socket> <pid_file>`` (the systemd template passes ``/%I`` first), so the
    match is the argv ELEMENT run ``-m <module> <repo_root>``: the module name
    directly after ``-m`` and the resolved repo root directly after it. A
    substring match is not identity (any process that merely mentions the
    module in an argument would qualify), and a daemon serving a DIFFERENT
    repo is not ours to claim or signal; an unreadable ``/proc`` entry (the
    pid is gone) is not a daemon. A RELATIVE repo root in the argv is the
    daemon's own cwd's business, so it resolves against ``/proc/<pid>/cwd``,
    never against the caller's cwd; the cwd link is read ONLY then (an
    absolute root needs no cwd), and an unreadable cwd link on a relative
    root is the same fail-safe ``False``.
    """
    try:
        raw = Path(f"/proc/{pid}/cmdline").read_bytes()
    except OSError:
        return False
    argv = [a for a in raw.split(b"\0") if a]
    module = _DAEMON_ENTRY_MODULE.encode()
    want = repo_root.resolve()
    for i in range(len(argv) - 2):
        if argv[i] == b"-m" and argv[i + 1] == module:
            root = Path(os.fsdecode(argv[i + 2]))
            if not root.is_absolute():
                try:
                    root = Path(os.readlink(f"/proc/{pid}/cwd")) / root
                except OSError:
                    return False
            return root.resolve() == want
    return False


@traces(spec="110", story="US-110-013", reqs=["REQ-110-135"])
def _process_alive(pid: int) -> bool:
    """Check if a process with the given PID is alive.

    A zombie (exited but not yet reaped) counts as dead: os.kill(pid, 0)
    succeeds on zombies, which previously made stop() wait out the full
    shutdown timeout and SIGKILL a process that had already exited
    cleanly. If the zombie is our own child, reap it.
    """
    try:
        os.kill(pid, 0)
    except ProcessLookupError, PermissionError:
        return False
    try:
        stat = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
        # State is the first field after the closing paren of comm (which
        # may itself contain spaces or parens).
        state = stat[stat.rindex(")") + 2]
    except OSError, ValueError, IndexError:
        return True
    if state == "Z":
        try:
            os.waitpid(pid, os.WNOHANG)
        except ChildProcessError, OSError:
            pass  # Not our child; nothing to reap.
        return False
    return True