feat(trainer): pod resource samples + vanish post-mortem — prove restart vs leak

cascade/trainer/remote.py

@@ -559,12 +559,109 @@ def _progress_prefix(run_dir: str) -> str:
     return f"{PROGRESS_FILE_ENV}={shlex.quote(detached_progress_path(run_dir))} "
 
 
+# 2026-11-13: three legs died with the pod dark 6-15 min and back with its run
+# dir GONE (empty log tails) at steps 14400-16450 — container recreated, or a
+# leak of ours killing it? Every poll now reads the pod's resources (host
+# MemAvailable, the container cgroup's memory use/limit, free disk in the
+# workdir, kernel uptime, pid 1's start) on one line before the token, so the
+# lead-up to a death is on record and a restart/reboot is PROVEN, not guessed.
+DETACHED_RESOURCES_MARK = "RES:"
+# fields, in order: mem_avail_kb cg_mem_bytes cg_max uptime_s pid1_start_ticks disk_free_kb
+_RESOURCES_SHELL = (
+    "printf '" + DETACHED_RESOURCES_MARK + "%s %s %s %s %s %s\\n' "
+    "\"$(awk '/^MemAvailable:/{print $2}' /proc/meminfo 2>/dev/null)\" "
+    "\"$(cat /sys/fs/cgroup/memory.current 2>/dev/null"
+    " || cat /sys/fs/cgroup/memory/memory.usage_in_bytes 2>/dev/null)\" "
+    "\"$(cat /sys/fs/cgroup/memory.max 2>/dev/null"
+    " || cat /sys/fs/cgroup/memory/memory.limit_in_bytes 2>/dev/null)\" "
+    "\"$(cut -d' ' -f1 /proc/uptime 2>/dev/null)\" "
+    "\"$(awk '{print $22}' /proc/1/stat 2>/dev/null)\" "
+    "\"$(df -Pk . 2>/dev/null | awk 'NR==2{print $4}')\"; "
+)
+# Resource samples kept per leg (one per poll: 30 s × 120 = the last hour);
+# one INFO line of them every RESOURCE_LOG_SECONDS while the worker runs.
+RESOURCE_HISTORY = 120
+RESOURCE_LOG_SECONDS = 1800
+
+
+def parse_resources(lines) -> dict | None:
+    """The resource record a poll printed (:data:`DETACHED_RESOURCES_MARK`),
+    or ``None``. Missing / unreadable fields are ``None``; never raises."""
+    names = ("mem_avail_kb", "cg_mem_bytes", "cg_max", "uptime_s", "pid1_start", "disk_free_kb")
+    for ln in lines:
+        if not ln.startswith(DETACHED_RESOURCES_MARK):
+            continue
+        vals = ln[len(DETACHED_RESOURCES_MARK):].split(" ")
+        out: dict = {}
+        for name, raw in zip(names, vals + [""] * len(names), strict=False):
+            raw = raw.strip()
+            if name == "cg_max" and raw == "max":
+                out[name] = "max"
+                continue
+            try:
+                out[name] = float(raw) if name == "uptime_s" else int(raw)
+            except ValueError:
+                out[name] = None
+        return out
+    return None
+
+
+def resource_verdict(history: list[dict]) -> str:
+    """What the resource samples PROVE about a death (``""`` = nothing):
+    pid 1 started again (container recreated) or the kernel uptime went
+    backwards (host rebooted) between the first and the last sample."""
+    seen = [r for r in history if r]
+    if len(seen) < 2:
+        return ""
+    first, last = seen[0], seen[-1]
+    up0, up1 = first.get("uptime_s"), last.get("uptime_s")
+    if up0 is not None and up1 is not None and up1 < up0:
+        return (f"HOST REBOOTED (kernel uptime {up0:.0f}s → {up1:.0f}s)")
+    p0, p1 = first.get("pid1_start"), last.get("pid1_start")
+    if p0 is not None and p1 is not None and p1 != p0:
+        return f"CONTAINER RESTARTED (pid 1 start {p0} → {p1} ticks after boot)"
+    return ""
+
+
+def _fmt_resources(r: dict | None) -> str:
+    if not r:
+        return "n/a"
+
+    def gib(v, scale):
+        return "?" if not isinstance(v, int) else f"{v * scale / 2**30:.1f}G"
+
+    lim = r.get("cg_max")
+    lim_s = "max" if lim == "max" else gib(lim, 1)
+    up = r.get("uptime_s")
+    return (f"mem_avail={gib(r.get('mem_avail_kb'), 1024)} cg_mem={gib(r.get('cg_mem_bytes'), 1)}"
+            f"/{lim_s} disk_free={gib(r.get('disk_free_kb'), 1024)} "
+            f"uptime={'?' if up is None else f'{up:.0f}s'} pid1_start={r.get('pid1_start')}")
+
+
+def _detached_postmortem_command(host: RemoteHost, run_dir: str) -> str:
+    """Read-only facts from a pod whose worker vanished, taken BEFORE teardown:
+    resources, boot / pid-1 times, whether the run dir survived, the kernel's
+    OOM lines when readable. Best-effort; every step tolerates failure."""
+    rd = shlex.quote(run_dir)
+    return (f"cd {shlex.quote(host.workdir)} 2>/dev/null; " + _RESOURCES_SHELL
+            + "echo \"boot: $(uptime -s 2>/dev/null)  now: $(date -u +%FT%TZ)\"; "
+            "echo \"pid1: $(ps -o lstart= -p 1 2>/dev/null) $(cat /proc/1/comm 2>/dev/null)\"; "
+            f"if [ -d {rd} ]; then echo \"run dir: present\"; ls -la {rd} 2>/dev/null | head -20; "
+            "else echo \"run dir: MISSING\"; fi; "
+            f"ls -la --time-style=+%FT%T {shlex.quote(DETACHED_RUN_ROOT)} 2>/dev/null | head -10; "
+            "(dmesg -T 2>/dev/null || journalctl -k --no-pager 2>/dev/null) "
+            "| grep -iE 'oom|killed process|out of memory|nvrm|xid' | tail -20; "
+            "free -m 2>/dev/null; df -h . /dev/shm 2>/dev/null; true")
+
+
 def _detached_poll_command(host: RemoteHost, run_dir: str) -> str:
     rd = shlex.quote(run_dir)
     pf = shlex.quote(detached_progress_path(run_dir))
     # A RUNNING worker also prints its progress record (when it writes one)
-    # on the line BEFORE the token — the token stays the last line.
-    return (f"cd {shlex.quote(host.workdir)} && if [ -f {rd}/exit_code ]; then "
+    # on the line BEFORE the token — the token stays the last line. Every poll
+    # leads with the pod's resource line (:data:`_RESOURCES_SHELL`).
+    return (f"cd {shlex.quote(host.workdir)} && " + _RESOURCES_SHELL
+            + f"if [ -f {rd}/exit_code ]; then "
             f"echo EXIT:$(cat {rd}/exit_code); elif [ -f {rd}/pid ] && "
             f"kill -0 $(cat {rd}/pid) 2>/dev/null; then "
             f"[ -f {pf} ] && printf '{DETACHED_PROGRESS_MARK}%s\\n' "
@@ -944,6 +1041,8 @@ def run_detached(host: RemoteHost, body: str, stdin_env: str | None, run_dir: st
     rc: int | None = None
     vanished = False
     last_progress: dict | None = None    # the worker's last step record seen by a poll
+    resources: list[dict] = []           # the pod's resource samples (parse_resources)
+    res_logged = 0.0
     dark_quarantined = False
     poll_argv = build_ssh_argv(host, _detached_poll_command(host, run_dir))
     while True:
@@ -1009,8 +1108,19 @@ def run_detached(host: RemoteHost, body: str, stdin_env: str | None, run_dir: st
                             "grace %ds) — the detached worker keeps running",
                             describe, misses, gap, grace_seconds)
             continue
+        res = parse_resources(lines)
+        if res is not None:
+            res["t"] = round(now, 1)
+            resources.append(res)
+            del resources[:-RESOURCE_HISTORY]
         if misses:
-            log.info("%s: pod reachable again after %d failed poll(s)", describe, misses)
+            log.info("%s: pod reachable again after %d failed poll(s) — %s%s", describe, misses,
+                     _fmt_resources(res),
+                     (f" — {resource_verdict(resources)}" if resource_verdict(resources) else ""))
+        elif res is not None and now - res_logged >= RESOURCE_LOG_SECONDS:
+            res_logged = now
+            log.info("%s: pod resources (step %s): %s", describe,
+                     (last_progress or {}).get("step"), _fmt_resources(res))
         misses = 0
         last_seen = now
         if token == "RUNNING":
@@ -1030,9 +1140,11 @@ def run_detached(host: RemoteHost, body: str, stdin_env: str | None, run_dir: st
             vanished = True
             rc = 1
             step = (last_progress or {}).get("step")
+            verdict = resource_verdict(resources)
             _quarantine_run_host(host, describe, "detached worker vanished mid-run without "
                                  "an exit code" + (f" (last step {step})" if step is not None
-                                                   else ""))
+                                                   else "")
+                                 + (f" — {verdict}" if verdict else ""))
             break
         try:
             rc = int(token.split(":", 1)[1])
@@ -1059,11 +1171,31 @@ def run_detached(host: RemoteHost, body: str, stdin_env: str | None, run_dir: st
     stdout, _, stderr = out.partition(DETACHED_STDERR_MARK)
     stderr, _, fetched = stderr.partition(DETACHED_FETCH_PROGRESS_MARK)
     progress = _parse_fetched_progress(fetched) or last_progress
+    evidence = ""
     if vanished:
+        # read-only, best-effort, AFTER the log fetch (the pod is torn down
+        # seconds later): what happened to the box itself
+        postmortem = ""
+        try:
+            pm = run(build_ssh_argv(host, _detached_postmortem_command(host, run_dir)), 90, "")
+            postmortem = (pm.stdout or "")[-8000:]
+        except Exception as e:  # noqa: BLE001 — evidence only, never fail the leg on it
+            postmortem = f"(postmortem failed: {e})"
+        # The failure TEXT stays exactly as before: the fault classifier
+        # (faults.classify_*) reads it to decide infra vs miner billing, and a
+        # dmesg "Killed process" line must never move that decision. The
+        # evidence rides separately (``proc.postmortem``) into the saved log.
         stderr = stderr.rstrip() + "\n[detached worker process vanished without an exit code]"
+        verdict = resource_verdict(resources)
+        evidence = ("=== verdict: " + (verdict or "not proven by the resource samples") + " ===\n"
+                    "=== pod postmortem (after the log fetch, before teardown) ===\n"
+                    + postmortem.rstrip()
+                    + "\n=== pod resources, last " + str(len(resources)) + " poll(s) ===\n"
+                    + "\n".join(f"{r.get('t')} {_fmt_resources(r)}" for r in resources))
     proc = subprocess.CompletedProcess(args=fetch_argv, returncode=rc,
                                        stdout=stdout, stderr=stderr)
     proc.progress = progress          # the worker's last step record (or None)
+    proc.postmortem = evidence
     return proc
 
 
@@ -2230,7 +2362,9 @@ def _arm_prefetch(command: str, run_dir: str | None = None) -> str:
                 header={"role": role, "host": host.name, "address": f"{host.host}:{host.port}",
                         "base_seed": base_seed, "exit_code": proc.returncode,
                         "progress": progress},
-                stdout=proc.stdout or "", stderr=proc.stderr or "")
+                stdout=proc.stdout or "",
+                stderr=(proc.stderr or "") + (("\n" + proc.postmortem)
+                                              if getattr(proc, "postmortem", "") else ""))
             where = f" [evidence: {saved}]" if saved else ""
         if proc.returncode == 3:
             # Worker rc=3 = miner submission rejected (CorpusError): the

tests/unit/test_miner_cost_protections.py

@@ -230,7 +230,10 @@ def test_poll_command_prints_the_progress_record_before_the_token(tmp_path):
     host = RemoteHost(name="x", host="h", workdir=str(tmp_path))
     cmd = remote_mod._detached_poll_command(host, str(rd))
     out = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True).stdout
-    assert out.splitlines() == ["RUNNING"]                       # no file: unchanged
+    # no progress file: the resource line, then the token (2026-11-13)
+    assert out.splitlines()[-1] == "RUNNING" and len(out.splitlines()) == 2
+    assert out.splitlines()[0].startswith(remote_mod.DETACHED_RESOURCES_MARK)
+    assert parse_progress(out.splitlines()) is None
     (rd / "progress.json").write_text('{"step": 50,\n "tokens": 7}')
     out = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True).stdout
     lines = out.splitlines()

tests/unit/test_pod_vanish_evidence.py

@@ -0,0 +1,139 @@
+"""Pod resource samples + vanish post-mortem (2026-11-13).
+
+Three legs died with the pod dark 6-15 min and back with its run dir gone,
+at steps 14400-16450. Every detached poll now reads the pod's resources; a
+vanish runs a read-only post-mortem before teardown; the samples PROVE a
+container restart or a host reboot. The failure TEXT the fault classifier
+bills on is unchanged — evidence rides only into the saved log.
+"""
+from __future__ import annotations
+
+import subprocess
+
+import cascade.trainer.remote as R
+from cascade.trainer.remote import (
+    DETACHED_LAUNCH_TOKEN,
+    DETACHED_RESOURCES_MARK,
+    DETACHED_STDERR_MARK,
+    RemoteHost,
+    parse_resources,
+    resource_verdict,
+    run_detached,
+)
+
+
+def _proc(rc=0, stdout="", stderr=""):
+    return subprocess.CompletedProcess(args=[], returncode=rc, stdout=stdout, stderr=stderr)
+
+
+class _Scripted:
+    def __init__(self, steps):
+        self.steps = list(steps)
+        self.argvs = []
+
+    def __call__(self, argv, timeout, stdin):
+        self.argvs.append(argv)
+        step = self.steps.pop(0) if self.steps else _proc(stdout="RUNNING\n")
+        if isinstance(step, BaseException):
+            raise step
+        return step
+
+
+def _host():
+    return RemoteHost(name="p", host="h", workdir="/w")
+
+
+def _res(mem=500_000_000, cg=60 * 2**30, up=1000.0, pid1=4242, disk=900_000_000):
+    return f"{DETACHED_RESOURCES_MARK}{mem} {cg} max {up} {pid1} {disk}\n"
+
+
+def test_poll_command_emits_a_parseable_resource_line(tmp_path):
+    rd = tmp_path / "run"
+    rd.mkdir()
+    host = RemoteHost(name="x", host="h", workdir=str(tmp_path))
+    out = subprocess.run(["bash", "-c", R._detached_poll_command(host, str(rd))],
+                         capture_output=True, text=True).stdout.splitlines()
+    assert out[-1] == "GONE"                                   # no pid file
+    r = parse_resources(out)
+    assert r is not None and set(r) == {"mem_avail_kb", "cg_mem_bytes", "cg_max", "uptime_s",
+                                        "pid1_start", "disk_free_kb"}
+    assert isinstance(r["uptime_s"], float) and r["uptime_s"] > 0   # Linux CI
+    assert isinstance(r["disk_free_kb"], int)
+
+
+def test_parse_resources_tolerates_missing_fields():
+    assert parse_resources(["RUNNING"]) is None
+    r = parse_resources([f"{DETACHED_RESOURCES_MARK}123  max  "])
+    assert r["mem_avail_kb"] == 123 and r["cg_mem_bytes"] is None and r["cg_max"] == "max"
+    assert r["uptime_s"] is None and r["pid1_start"] is None and r["disk_free_kb"] is None
+
+
+def test_resource_verdict_proves_restart_and_reboot():
+    a = parse_resources([_res(up=5000.0, pid1=4242)])
+    assert resource_verdict([a]) == ""
+    assert resource_verdict([a, parse_resources([_res(up=5600.0, pid1=4242)])]) == ""
+    assert "CONTAINER RESTARTED" in resource_verdict(
+        [a, parse_resources([_res(up=5600.0, pid1=99999)])])
+    assert "HOST REBOOTED" in resource_verdict(
+        [a, parse_resources([_res(up=300.0, pid1=12)])])
+
+
+def test_vanish_runs_a_postmortem_and_keeps_the_classified_text_unchanged():
+    runner = _Scripted([
+        _proc(stdout=DETACHED_LAUNCH_TOKEN + "\n"),
+        _proc(stdout=_res(up=8000.0, pid1=4242) + "RUNNING\n"),
+        _proc(rc=255, stderr="ssh: timeout"),                    # dark
+        _proc(stdout=_res(up=8400.0, pid1=777777) + "GONE\n"),   # back, worker gone
+        _proc(stdout=DETACHED_STDERR_MARK + "\nlast log line\n"),  # log fetch
+        _proc(stdout="run dir: MISSING\n[ 1.0] Out of memory: Killed process 9 (python)\n"),
+    ])
+    proc = run_detached(_host(), "true", None, "/rd", timeout=600, poll_seconds=1,
+                        grace_seconds=600, runner=runner)
+    assert proc.returncode == 1
+    # the billed-on text: exactly the pre-2026-11-13 shape — no dmesg, no verdict
+    assert proc.stderr.rstrip().endswith(
+        "last log line\n[detached worker process vanished without an exit code]")
+    assert "Killed process" not in proc.stderr and "RESTARTED" not in proc.stderr
+    # the evidence carries the verdict, the postmortem and the samples
+    assert "CONTAINER RESTARTED" in proc.postmortem
+    assert "run dir: MISSING" in proc.postmortem and "Killed process" in proc.postmortem
+    assert "last 2 poll(s)" in proc.postmortem
+    # the postmortem ran AFTER the log fetch
+    cmds = [" ".join(a) for a in runner.argvs]
+    assert "run dir: present" in cmds[-1] and "tail -c" in cmds[-2]
+
+
+def test_postmortem_failure_never_fails_differently():
+    runner = _Scripted([
+        _proc(stdout=DETACHED_LAUNCH_TOKEN + "\n"),
+        _proc(stdout="GONE\n"),
+        _proc(stdout=DETACHED_STDERR_MARK + "\nx\n"),
+        OSError("ssh died"),
+    ])
+    proc = run_detached(_host(), "true", None, "/rd", timeout=60, poll_seconds=1, runner=runner)
+    assert proc.returncode == 1 and "vanished" in proc.stderr
+    assert "postmortem failed: ssh died" in proc.postmortem
+    assert "not proven" in proc.postmortem
+
+
+def test_clean_exit_has_no_postmortem_and_no_extra_ssh():
+    runner = _Scripted([
+        _proc(stdout=DETACHED_LAUNCH_TOKEN + "\n"),
+        _proc(stdout=_res() + "RUNNING\n"),
+        _proc(stdout=_res() + "EXIT:0\n"),
+        _proc(stdout="out\n" + DETACHED_STDERR_MARK + "\n"),
+    ])
+    proc = run_detached(_host(), "true", None, "/rd", timeout=60, poll_seconds=1, runner=runner)
+    assert proc.returncode == 0 and proc.postmortem == ""
+    assert len(runner.argvs) == 4
+
+
+def test_evidence_file_gets_the_postmortem(tmp_path):
+    from cascade.trainer.remote import save_failed_leg_evidence
+
+    p = save_failed_leg_evidence(tmp_path, "5Ctyc5w8gwif", header={"exit_code": 1},
+                                 stdout="", stderr="tail\n=== verdict: X ===\nrun dir: MISSING")
+    from pathlib import Path
+
+    text = Path(p).read_text(encoding="utf-8")
+    assert "=== verdict: X ===" in text and "run dir: MISSING" in text