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
pid; when a server dies or its pid changes 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 collect_forensics(socket_path, last_pid):
    """Best-effort death evidence. Dict of strings, never raises."""
    ev = {"ts": _now(), "socket": socket_path, "last_pid": last_pid}
    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 evaluate(previous, probed):
    """Pure transition logic: (prev_state, {sock: pid|None}) ->
    (new_state, events). Events: death | restart | started."""
    new_state, events = {}, []
    for sock, pid in sorted(probed.items()):
        prev = (previous.get(sock) or {})
        prev_pid = prev.get("pid")
        if pid 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})
        else:
            new_state[sock] = {"pid": pid, "since": _now()}
            if prev_pid and prev_pid != pid:
                # Changed with no witnessed death: restart inside one
                # tick gap (or pid recycled under us). Treat as a
                # restart, still worth a forensics note.
                events.append({"type": "restart", "socket": sock,
                               "old_pid": prev_pid, "pid": pid})
            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(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"])
            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"])
            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())