rune/scripts/ensure_core.py

#!/usr/bin/env python3
"""ensure_core.py — chat-live P1 F2: the REAL core, supervised, on 8376.

The hazard this script kills (research/factory-chat-live 05-adversarial:
"stale stub owns the core port"): a leftover parity canary
(cs-parity-serve, active_model 'stub-a') can hold 8376 while the real
operator-backed core is down — every chat-live answer then silently
comes from the stub, which is the worst failure mode (looks live,
isn't).

Verbs:
  ensure   (default) one supervised core on 8376, health-gated:
             real+healthy          -> ok, nothing to do
             stale/stub holder     -> kill it, start the real core
             port free             -> start the real core
           gate: /v1/status answers identity 'Rune' with a non-stub
           active_model and a real model listing; ~60s startup budget.
  status   report who holds the port and whether it passes the gate
           ('health-check' is an alias)
  stop     stop the core this script started (pidfile) — never a foreign
           process it did not identify as stale

Supervision target (wave-core-supervision-go-gc): the Go core
(~/.local/share/rune/bin/rune-core, source src/go/rune-core — same
endpoints, same identity-'Rune' gate) is the DEFAULT; RUNE_CORE=python
is the explicit fallback to the legacy core_server.py path. Each mode
keeps its own pidfile (rune_core_go.pid / core_server.pid) under
<rune_home>/logs.

The child is detached (its own session) with stdout/stderr to
<home>/logs/core_server.log — it survives this script's exit; that IS
the supervision contract for phase 1 (ensure + gate + stale-kill;
restart-on-death is a later wave).
"""
from __future__ import annotations

import json
import os
import signal
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]
PORT = 8376
GATE_TIMEOUT = 60.0


def rune_home() -> Path:
    env = os.environ.get("RUNE_HOME")
    if env:
        return Path(env)
    return Path.home() / ".rune"


def PY_CORE() -> Path:
    return REPO / "src" / "rune" / "core_server.py"


def GO_CORE() -> Path:
    env = os.environ.get("RUNE_CORE_BIN")
    if env:
        return Path(env)
    return Path.home() / ".local" / "share" / "rune" / "bin" / "rune-core"


def core_mode() -> str:
    mode = os.environ.get("RUNE_CORE", "go").strip().lower()
    if mode not in ("go", "python"):
        print(f"RUNE_CORE={mode!r} not understood (go|python) — using go")
        mode = "go"
    return mode

def deploy_tool_executor() -> Path | None:
    """D57 (Gitea #2590/#2591): the tool-executor binary is DEPLOYED to
    ~/.local/share/rune/bin alongside rune-core — anything shelling to
    it resolves that location via the core's runtime-env layer. Builds
    it from src/go/tool-executor when missing (never a silent failure:
    the outcome is printed and returned)."""
    src = REPO / "src" / "go" / "tool-executor"
    bindir = Path.home() / ".local" / "share" / "rune" / "bin"
    dest = bindir / "tool-executor"
    if dest.is_file() and os.access(dest, os.X_OK):
        return dest
    if not src.is_dir():
        print(f"  tool-executor source missing: {src} — not deployed")
        return None
    try:
        bindir.mkdir(parents=True, exist_ok=True)
        r = subprocess.run(
            ["go", "build", "-o", str(dest), "."],
            cwd=src, capture_output=True, text=True, timeout=300,
        )
        if r.returncode != 0:
            print(f"  tool-executor build FAILED: {r.stderr.strip()[-400:]}")
            return None
        print(f"  deployed tool-executor -> {dest}")
        return dest
    except (OSError, subprocess.SubprocessError) as exc:
        print(f"  tool-executor deploy failed: {exc}")
        return None


def core_target() -> Path:
    return PY_CORE() if core_mode() == "python" else GO_CORE()


def pidfile_for(mode: str) -> Path:
    name = "core_server.pid" if mode == "python" else "rune_core_go.pid"
    return rune_home() / "logs" / name


def _get(base: str, path: str, timeout: float = 3.0):
    try:
        with urllib.request.urlopen(base + path, timeout=timeout) as r:
            if r.status == 200:
                return json.loads(r.read().decode())
    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError):
        return None
    return None


def status_of(base: str):
    return _get(base, "/v1/status")


def models_of(base: str):
    body = _get(base, "/v1/models")
    if not body:
        return []
    return [m.get("id", "") for m in body.get("data", [])]


def gate_ok(base: str) -> tuple[bool, str]:
    """The F2 sanity gate: identity + active_model + real models."""
    st = status_of(base)
    if st is None:
        return False, "no /v1/status answer"
    ident = str(st.get("identity", ""))
    model = str(st.get("active_model", "") or "")
    if ident != "Rune":
        return False, f"identity {ident!r} (stale stub or foreign holder)"
    if model.startswith("stub"):
        return False, f"active_model {model!r} is a stub"
    ids = models_of(base)
    if not ids:
        return False, "no models listed (real core must expose its registry)"
    if any(i.startswith("stub") for i in ids):
        return False, f"stub models in the listing: {ids!r}"
    return True, f"identity {ident!r}, active_model {model!r}, {len(ids)} models"


def port_holder(port: int) -> int | None:
    """The pid LISTENING on port (pure /proc walk — no external tools)."""
    want = f"{port:04X}"
    inodes = set()
    for fname in ("/proc/net/tcp", "/proc/net/tcp6"):
        try:
            lines = Path(fname).read_text().splitlines()[1:]
        except OSError:
            continue
        for ln in lines:
            parts = ln.split()
            if len(parts) < 10 or parts[1].split(":")[-1].upper() != want:
                continue
            if parts[3] == "0A":  # LISTEN
                inodes.add(parts[9])
    if not inodes:
        return None
    for proc in Path("/proc").iterdir():
        if not proc.name.isdigit():
            continue
        try:
            for fd in (proc / "fd").iterdir():
                try:
                    target = fd.readlink()
                    s = target.name if hasattr(target, "name") else str(target)
                    if s.startswith("socket:[") and s[len("socket:["):-1] in inodes:
                        return int(proc.name)
                except OSError:
                    continue
        except OSError:
            continue
    return None


def pid_of(holder: int) -> str:
    try:
        return Path(f"/proc/{holder}/comm").read_text().strip()
    except OSError:
        return "?"


def kill_holder(holder: int, name: str) -> bool:
    print(f"  killing stale holder pid {holder} ({name})")
    try:
        os.kill(holder, signal.SIGTERM)
        for _ in range(30):
            if Path(f"/proc/{holder}").exists() is False:
                return True
            time.sleep(0.1)
        os.kill(holder, signal.SIGKILL)
        time.sleep(0.3)
        return not Path(f"/proc/{holder}").exists()
    except OSError as exc:
        print(f"  kill failed: {exc}")
        return False


def start_core(port: int) -> tuple[int | None, Path]:
    home = rune_home()
    logs = home / "logs"
    logs.mkdir(parents=True, exist_ok=True)
    mode = core_mode()
    if mode == "go":
        deploy_tool_executor()
        binary = GO_CORE()
        if not binary.is_file() or not os.access(binary, os.X_OK):
            print(f"GO CORE MISSING: {binary} — build it (scripts/"
                  f"go_build_sweep.py) or set RUNE_CORE=python; not "
                  f"falling back silently")
            return None, logs / "rune_core_go.log"
        argv = [str(binary), "--host", "127.0.0.1", "--port", str(port)]
        log_path = logs / "rune_core_go.log"
    else:
        argv = [sys.executable, str(PY_CORE()),
                "--host", "127.0.0.1", "--port", str(port)]
        log_path = logs / "core_server.log"
    log = open(log_path, "ab")  # noqa: SIM115 — detached child owns it now
    proc = subprocess.Popen(
        argv,
        stdout=log, stderr=subprocess.STDOUT,
        start_new_session=True,  # detached: survives this script
    )
    pidfile_for(mode).write_text(f"{proc.pid}\n")
    print(f"  started {mode} core pid {proc.pid} (log {log_path})")
    return proc.pid, log_path


def wait_gate(base: str, timeout: float = GATE_TIMEOUT) -> tuple[bool, str]:
    deadline = time.time() + timeout
    last = "gate timeout"
    while time.time() < deadline:
        ok, detail = gate_ok(base)
        if ok:
            return True, detail
        last = detail
        time.sleep(0.5)
    return False, last


def running_binary_fresh(holder: int) -> tuple[bool, str]:
    """#2638 (deploy-divergence hardening): is the RUNNING core the
    binary currently on disk? Compares /proc/<pid>/exe against the
    supervision target's path and mtime — never file mtimes alone, the
    path identity comes first. RUNE_CORE_FRESHNESS_OFF=1 disables."""
    if os.environ.get("RUNE_CORE_FRESHNESS_OFF") == "1":
        return True, "freshness check disabled (RUNE_CORE_FRESHNESS_OFF=1)"
    try:
        exe = Path(f"/proc/{holder}/exe").resolve()
    except OSError:
        return False, "running binary unreadable (/proc/<pid>/exe)"
    disk = core_target()
    if not disk.is_file():
        return False, f"on-disk binary missing: {disk}"
    if exe != disk.resolve():
        return False, f"running binary {exe} != on-disk {disk}"
    try:
        run_mtime = os.stat(f"/proc/{holder}/exe").st_mtime
        disk_mtime = disk.stat().st_mtime
    except OSError as exc:
        return False, f"mtime comparison failed: {exc}"
    if disk_mtime > run_mtime + 1.0:
        return False, "on-disk binary is newer than the running one"
    return True, f"binary fresh ({disk})"


def restart_core_onto_fresh_binary(holder: int, why: str) -> int:
    """#2638: a real, healthy core serving from a STALE binary is the
    divergence failure mode — restart it onto the on-disk binary. The
    owner ruled dominion/no-asking, so the restart is performed (not
    offered) and logged loudly. Returns the ensure verdict (0/1)."""
    print(f"RESTARTING core onto the fresh binary (#2638 freshness law): {why}")
    if not kill_holder(holder, pid_of(holder)):
        print("FRESHNESS RESTART FAILED: could not stop the stale core")
        return 1
    time.sleep(0.3)
    pid, _ = start_core(PORT)
    if pid is None:
        return 1
    ok, detail = wait_gate(f"http://127.0.0.1:{PORT}")
    if ok:
        print(f"RESTARTED onto fresh binary — REAL CORE LIVE on {PORT}: {detail}")
        return 0
    print(f"GATE FAILED after restart: {detail}")
    return 1


def main(argv: list[str]) -> int:
    verb = argv[0] if argv else "ensure"
    base = f"http://127.0.0.1:{PORT}"

    if verb in ("status", "health-check"):
        holder = port_holder(PORT)
        if holder is None:
            print(f"port {PORT}: free")
            return 1
        ok, detail = gate_ok(base)
        mode = core_mode()
        mode_tag = ""
        if pidfile_for(mode).exists():
            mode_tag = f" [{mode}, pidfile {pidfile_for(mode).name}]"
        print(f"port {PORT}: pid {holder} ({pid_of(holder)}) — "
              f"{'HEALTHY' if ok else 'STALE'}: {detail}{mode_tag}")
        return 0 if ok else 1

    if verb == "stop":
        mode = core_mode()
        pidfile = pidfile_for(mode)
        if not pidfile.exists() and mode == "go":
            pidfile = rune_home() / "logs" / "core_server.pid"
        if not pidfile.exists():
            print("no pidfile — nothing this script started")
            return 1
        pid = int(pidfile.read_text().strip())
        try:
            os.kill(pid, signal.SIGTERM)
            print(f"stopped {mode} core pid {pid}")
            pidfile.unlink()
            return 0
        except OSError as exc:
            print(f"stop failed: {exc}")
            return 1

    if verb != "ensure":
        print(__doc__)
        return 2

    ok, detail = gate_ok(base)
    holder = port_holder(PORT)
    st = status_of(base)
    is_rune = st is not None and str(st.get("identity", "")) == "Rune"

    if ok and holder is not None and is_rune:
        fresh, why = running_binary_fresh(holder)
        if not fresh:
            return restart_core_onto_fresh_binary(holder, why)

    if ok:
        print(f"core already real and healthy — {detail}")
        return 0

    if holder is not None:
        name = pid_of(holder)
        if is_rune:
            # a REAL core that fails the gate for another reason (sick,
            # model-less): if its binary is stale, the #2638 freshness
            # law still restarts it onto the on-disk binary
            fresh, why = running_binary_fresh(holder)
            if not fresh:
                return restart_core_onto_fresh_binary(holder,
                                                      f"gate failing + stale binary: {why}")
            # report, do not kill — never kill what is real and fresh
            print(f"real core pid {holder} fails the gate ({detail}) — NOT killing")
            return 1
        if not kill_holder(holder, name):
            return 1
    time.sleep(0.3)

    pid, _ = start_core(PORT)
    if pid is None:
        return 1
    ok, detail = wait_gate(base)
    if not ok:
        print(f"GATE FAILED after {GATE_TIMEOUT:.0f}s: {detail}")
        return 1
    print(f"REAL CORE LIVE on {PORT} — {detail}")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))