宜博 / lemongraph

"""Deploy Supervisor process skeleton (U-036).

Independent OS process (started via `python3 -m supervisor.deploy_supervisor`),
pure stdlib, no LLM CLI dependency -- per docs/done/supervisor-auto-rollback-design.md
§2's "three independent processes, no shared failure domain" principle:
Deploy Supervisor crashing must not affect Live/Candidate, and Live/Candidate
crashing must not affect Deploy Supervisor's own ability to keep observing
and act.

PID/status file conventions adapted (NOT imported -- see CLAUDE.md's "do not
import graphloop code directly" rule) from graphloop's hermes_watchdog.py:
atomic writes (tempfile-in-same-dir + os.replace), liveness via
`os.kill(pid, 0)`, fail-closed (this skeleton never auto-restarts anything --
a crashing `poll_fn` stops the loop and marks status='stopped' rather than
silently dying while still holding the PID file, which would make
`is_running()` lie).

Scope: this Unit owns process lifecycle only (start/stop/status/PID file).
The actual "check for staged deployments and act" logic is U-037+'s job,
injected here as the `poll_fn` callback -- U-036 has no opinion about what
that logic does, only about running it on a schedule and surviving/reporting
its own crashes correctly.
"""

import argparse
import json
import os
import signal
import subprocess
import sys
import time

from src import paths


def runtime_dir():
    d = os.path.join(paths.platform_home(), "deploy_supervisor")
    os.makedirs(d, exist_ok=True)
    return d


def pid_file_path():
    return os.path.join(runtime_dir(), "supervisor.pid")


def status_file_path():
    return os.path.join(runtime_dir(), "supervisor.status.json")


def _atomic_write_json(path, data):
    tmp = path + ".{}.tmp".format(os.getpid())
    with open(tmp, "w", encoding="utf-8") as handle:
        json.dump(data, handle)
    os.replace(tmp, path)


def is_alive(pid):
    """`os.kill(pid, 0)`-based liveness -- NOTE this can't distinguish "truly
    running" from "exited but not yet reaped by its parent" (a zombie's PID
    slot still answers kill(pid, 0) successfully until something calls
    wait() on it). If the calling process spawned `pid` itself via
    `subprocess.Popen` and holds onto it across multiple `is_alive()` calls
    without ever waiting on it, this can report True forever for an
    already-dead process -- found for real (2026-08-02, U-039) via a test
    that actually waited for a stopped process to go away and it never did.
    Callers that spawned `pid` themselves should use `reap_if_exited()`
    instead, which handles this correctly; this function is still right for
    checking a pid this process did NOT spawn (nothing to reap, this is the
    best available check)."""
    if not isinstance(pid, int) or pid <= 0:
        return False
    try:
        os.kill(pid, 0)
    except OSError:
        return False
    return True


def reap_if_exited(pid):
    """True once `pid` is confirmed gone. If `pid` is a real child of the
    calling process (e.g. spawned via `subprocess.Popen` earlier in this
    same process's lifetime), this actually reaps it via a non-blocking
    `os.waitpid` -- required for `is_alive()` to ever correctly report False
    for it afterward (see that function's own docstring for why). If `pid`
    isn't our child (belongs to a different process/session -- e.g. a
    `live_pid` inherited from a prior Deploy Supervisor run), `os.waitpid`
    raises `ChildProcessError` and this falls back to `is_alive()`'s plain
    liveness check instead, which is the best available answer without
    being the real parent."""
    try:
        reaped_pid, _ = os.waitpid(pid, os.WNOHANG)
        if reaped_pid == pid:
            return True
    except ChildProcessError:
        pass
    return not is_alive(pid)


def write_status(status, detail=None):
    _atomic_write_json(status_file_path(), {
        "status": status, "pid": os.getpid(), "updated_at": time.time(), "detail": detail,
    })


def read_status():
    try:
        with open(status_file_path(), encoding="utf-8") as handle:
            return json.load(handle)
    except (OSError, json.JSONDecodeError):
        return None


def is_running():
    """True if a Deploy Supervisor process is currently alive per the PID
    file. Doesn't distinguish "never started" from "PID file stale" --
    callers needing that distinction should also check `read_status()`."""
    try:
        with open(pid_file_path(), encoding="utf-8") as handle:
            pid = int(handle.read().strip())
    except (OSError, ValueError):
        return False
    return is_alive(pid)


class _StopRequested(Exception):
    """Internal: raised by the SIGTERM handler to break run_foreground's loop."""


def run_foreground(poll_fn=None, poll_interval=5.0, iterations=None):
    """Blocking main loop -- calls `poll_fn()` (if given) once per
    `poll_interval` seconds, up to `iterations` cycles (None = forever, only
    exits on SIGTERM). Writes a PID file + 'running' status on entry, and
    unconditionally writes a 'stopped' status + removes the PID file on any
    exit path (normal completion, SIGTERM, or an unhandled exception from
    `poll_fn`) -- an exception from `poll_fn` propagates to the caller after
    cleanup runs, it is not swallowed (fail-closed: a broken poll cycle
    should surface, not vanish into an infinite silent retry).
    """
    def _handle_sigterm(signum, frame):
        raise _StopRequested()

    previous_handler = signal.signal(signal.SIGTERM, _handle_sigterm)
    with open(pid_file_path(), "w", encoding="utf-8") as handle:
        handle.write(str(os.getpid()))
    write_status("running")
    try:
        count = 0
        while iterations is None or count < iterations:
            if poll_fn is not None:
                poll_fn()
            count += 1
            if iterations is not None and count >= iterations:
                break
            time.sleep(poll_interval)
    except _StopRequested:
        pass
    finally:
        signal.signal(signal.SIGTERM, previous_handler)
        write_status("stopped")
        if os.path.exists(pid_file_path()):
            os.remove(pid_file_path())


def start_detached(poll_interval=5.0):
    """Launches `python3 -m supervisor.deploy_supervisor` as a real,
    independent OS process (Popen, not a thread) -- crashing does not take
    down whatever called this. Returns the child's pid. Raises RuntimeError
    if a Deploy Supervisor is already running (per its own PID file)."""
    if is_running():
        raise RuntimeError("Deploy Supervisor is already running")
    process = subprocess.Popen(
        [sys.executable, "-m", "supervisor.deploy_supervisor", "--poll-interval", str(poll_interval)],
        cwd=paths.repo_root(), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )
    return process.pid


def stop(timeout=10.0):
    """Sends SIGTERM to the running Deploy Supervisor (if any) and waits up
    to `timeout` seconds for it to exit cleanly (PID file removed). Returns
    True if it stopped, False if it was already not running or didn't stop
    in time -- fail-closed: this does not SIGKILL on timeout, that's a
    judgment call left to whoever's driving it, not silently forced here."""
    try:
        with open(pid_file_path(), encoding="utf-8") as handle:
            pid = int(handle.read().strip())
    except (OSError, ValueError):
        return False
    if not is_alive(pid):
        return False
    os.kill(pid, signal.SIGTERM)
    deadline = time.time() + timeout
    while time.time() < deadline:
        if not is_running():
            reap_if_exited(pid)  # best-effort zombie cleanup, doesn't affect the True/False result
            return True
        time.sleep(0.1)
    return False


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--poll-interval", type=float, default=5.0)
    args = parser.parse_args()
    run_foreground(poll_interval=args.poll_interval)


if __name__ == "__main__":
    main()