宜博 / 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/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):
    if not isinstance(pid, int) or pid <= 0:
        return False
    try:
        os.kill(pid, 0)
    except OSError:
        return False
    return True


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():
            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()