diff --git a/cascade/trainer/remote.py b/cascade/trainer/remote.py index 4291d0cc..37103ad4 100644 --- a/cascade/trainer/remote.py +++ b/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 diff --git a/tests/unit/test_miner_cost_protections.py b/tests/unit/test_miner_cost_protections.py index b1ffd7ad..a2fdd38a 100644 --- a/tests/unit/test_miner_cost_protections.py +++ b/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() diff --git a/tests/unit/test_pod_vanish_evidence.py b/tests/unit/test_pod_vanish_evidence.py new file mode 100644 index 00000000..7db1f90b --- /dev/null +++ b/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