box/bin/tmux_server_watchdog.py

#!/usr/bin/env python3
"""tmux_server_watchdog.py — Death-capture for tmux servers.

Runs on a 1-minute systemd timer. Remembers each known socket's server
identity (pid + /proc starttime + ppid + cmdline); when a server dies,
its pid changes, or its pid is recycled under us without a witnessed
death, appends a forensics bundle (dmesg OOM/kill lines, memory,
uptime, journal tail) to logs/tmux-server-deaths.jsonl so the next
"tmux crashed" leaves evidence instead of a mystery.

Read-only against tmux itself: one `display-message -p` probe per
socket. Never raises; a watchdog must not need its own watchdog.
"""

import json
import os
import subprocess
import sys
from datetime import datetime, timezone

BIN_DIR = os.path.dirname(os.path.abspath(__file__))
REPO_ROOT = os.path.dirname(BIN_DIR)
sys.path.insert(0, BIN_DIR)

try:
    from muse_choice_watcher import KNOWN_SOCKETS
except Exception:
    KNOWN_SOCKETS = ["/tmp/tmux-1000/default"]

STATE_FILE = os.path.join(REPO_ROOT, ".state", "tmux-servers.json")
DEATH_LOG = os.path.join(REPO_ROOT, "logs", "tmux-server-deaths.jsonl")


def _now():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _run(cmd, timeout=10):
    try:
        r = subprocess.run(cmd, capture_output=True, text=True,
                           timeout=timeout)
        return r.returncode, (r.stdout or "").strip()
    except Exception as e:
        return -1, "exec failed: %r" % (e,)


def probe(socket_path):
    """Server pid for a socket, or None when unreachable."""
    rc, out = _run(["tmux", "-S", socket_path, "display-message",
                    "-p", "#{pid}"], timeout=10)
    if rc != 0:
        return None
    try:
        return int(out.strip().split()[0])
    except (ValueError, IndexError):
        return None


def proc_identity(pid):
    """Identity dict for a pid: starttime defeats PID-reuse confusion.

    Never raises; on any failure returns {"pid": pid} so callers can
    still snapshot. starttime is the raw /proc starttime tick (field
    22), stable for the life of the process."""
    ident = {"pid": pid}
    try:
        with open("/proc/%d/stat" % pid) as f:
            parts = f.read().rsplit(")", 1)[1].split()
        # After "(comm)": state ppid pgrp session tty_nr ... starttime
        # is field 22 overall, i.e. parts[19] after the split above.
        ident["ppid"] = int(parts[1])
        ident["starttime"] = int(parts[19])
    except Exception:
        pass
    try:
        with open("/proc/%d/cmdline" % pid, "rb") as f:
            raw = f.read().replace(b"\0", b" ").decode(
                "utf-8", "replace").strip()
        if raw:
            ident["cmd"] = raw[:200]
    except Exception:
        pass
    return ident


def probe_identity(socket_path):
    """Enriched snapshot for a socket: identity dict or None."""
    pid = probe(socket_path)
    if pid is None:
        return None
    return proc_identity(pid)


def collect_forensics(socket_path, last_pid, last_identity=None):
    """Best-effort death evidence. Dict of strings, never raises."""
    ev = {"ts": _now(), "socket": socket_path, "last_pid": last_pid}
    if last_identity:
        ev["last_identity"] = last_identity
    rc, dmesg = _run(["dmesg"], timeout=10)
    if rc != 0:
        ev["dmesg"] = "unavailable: %s" % dmesg[:200]
    else:
        hits = [ln for ln in dmesg.split("\n")
                if any(k in ln.lower() for k in
                       ("oom", "killed process", "segfault", "tmux"))]
        ev["dmesg_hits"] = hits[-15:]
    _, ev["memory"] = _run(["free", "-m"], timeout=10)
    _, ev["uptime"] = _run(["uptime"], timeout=10)
    rc, journal = _run(["journalctl", "--user", "-n", "50"], timeout=10)
    if rc == 0:
        ev["journal_tmux"] = [ln for ln in journal.split("\n")
                              if "tmux" in ln.lower()][-10:]
    else:
        ev["journal_tmux"] = []
    return ev


def read_state(path=None):
    try:
        with open(path or STATE_FILE) as f:
            data = json.load(f)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def write_state(state, path=None):
    path = path or STATE_FILE
    try:
        parent = os.path.dirname(path)
        if parent:
            os.makedirs(parent, exist_ok=True)
        tmp = "%s.tmp.%d" % (path, os.getpid())
        with open(tmp, "w") as f:
            json.dump(state, f, indent=1)
        os.replace(tmp, path)
    except Exception:
        pass


def append_death(ev, path=None):
    path = path or DEATH_LOG
    try:
        parent = os.path.dirname(path)
        if parent:
            os.makedirs(parent, exist_ok=True)
        with open(path, "a") as f:
            f.write(json.dumps(ev) + "\n")
    except Exception:
        pass


def _as_identity(value):
    """Normalize a probed value to an identity dict (legacy int ok)."""
    if value is None:
        return None
    if isinstance(value, dict):
        return value
    return {"pid": value}


def _prev_identity(prev):
    ident = {"pid": prev.get("pid")}
    for key in ("starttime", "ppid", "cmd"):
        if prev.get(key) is not None:
            ident[key] = prev[key]
    return ident


def evaluate(previous, probed):
    """Pure transition logic: (prev_state, {sock: pid|identity|None}) ->
    (new_state, events). Events: death | restart | started.

    Probed values may be a bare pid (legacy) or an identity dict from
    probe_identity(). Same pid with a different starttime is a restart
    (pid recycled under us), not steady state."""
    new_state, events = {}, []
    for sock, raw in sorted(probed.items()):
        ident = _as_identity(raw)
        prev = (previous.get(sock) or {})
        prev_pid = prev.get("pid")
        if ident is None:
            new_state[sock] = {"pid": None, "died": _now(),
                               "last_pid": prev_pid}
            if prev_pid:
                events.append({"type": "death", "socket": sock,
                               "last_pid": prev_pid,
                               "last_identity": _prev_identity(prev)})
        else:
            pid = ident.get("pid")
            new_state[sock] = dict(ident, since=_now())
            if prev_pid and prev_pid != pid:
                # Changed with no witnessed death: restart inside one
                # tick gap. Worth a forensics note.
                events.append({"type": "restart", "socket": sock,
                               "old_pid": prev_pid, "pid": pid,
                               "last_identity": _prev_identity(prev)})
            elif (prev_pid and prev_pid == pid
                    and prev.get("starttime") is not None
                    and ident.get("starttime") is not None
                    and prev["starttime"] != ident["starttime"]):
                # Same pid, different process: pid recycled under us.
                events.append({"type": "restart", "socket": sock,
                               "old_pid": prev_pid, "pid": pid,
                               "pid_reused": True,
                               "last_identity": _prev_identity(prev)})
            elif not prev_pid and prev.get("died"):
                events.append({"type": "started", "socket": sock,
                               "pid": pid})
            elif not prev_pid and not prev:
                events.append({"type": "started", "socket": sock,
                               "pid": pid})
    return new_state, events


def check(sockets=None, dry_run=False):
    """Probe, transition state, log deaths. Returns summary dict."""
    probed = {s: probe_identity(s) for s in (sockets or KNOWN_SOCKETS)}
    previous = read_state()
    new_state, events = evaluate(previous, probed)
    for ev in events:
        if ev["type"] == "death":
            bundle = collect_forensics(ev["socket"], ev["last_pid"],
                                       ev.get("last_identity"))
            bundle["event"] = "death"
            if not dry_run:
                append_death(bundle)
            ev["forensics"] = bundle
        elif ev["type"] == "restart":
            bundle = collect_forensics(ev["socket"], ev["old_pid"],
                                       ev.get("last_identity"))
            bundle["event"] = "restart-gap-missed"
            if not dry_run:
                append_death(bundle)
            ev["forensics"] = bundle
    if not dry_run:
        write_state(new_state)
    return {"probed": probed, "events": events, "dry_run": dry_run}


def main(argv=None):
    import argparse
    ap = argparse.ArgumentParser(description="tmux server death-capture")
    ap.add_argument("--sockets", nargs="*", default=None)
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--json", action="store_true")
    args = ap.parse_args(argv)
    try:
        res = check(sockets=args.sockets, dry_run=args.dry_run)
    except Exception as e:
        print("watchdog failed: %r" % (e,), file=sys.stderr)
        return 1
    if args.json or args.dry_run:
        print(json.dumps(res, indent=1, default=str))
    else:
        for ev in res["events"]:
            print("%s: %s" % (ev["type"], ev["socket"]))
    return 0


if __name__ == "__main__":
    sys.exit(main())