"""Watch the box from the laptop. A different machine, a different failure domain. The heartbeat on the box catches a broken script. It cannot catch a dead box, a full disk, a network that went away, or a systemd timer that quietly stopped being enabled — because in every one of those cases the heartbeat is dead too. A watchman that shares a power supply with the thing it watches is not a watchman. So this runs on the laptop, reaches the box over SSH, and shouts through the laptop's own Telegram credentials. Nothing in the path is shared with the box except the network. **Its honest limit, stated rather than buried:** the laptop sleeps. If it is off for three days this does not run for three days. That is the whole trade — it adds no vendor and no new credential, and it covers the case that actually happens, which is the box being fine while the laptop is shut. Rocky opens the laptop most days. If he wants cover that is always-on, that is the external dead-man option in PLAN.md and it costs one small vendor. Reads only. Sends only to the one configured chat. Cannot touch the box's data. """ from __future__ import annotations import json import subprocess import time from datetime import datetime, timedelta, timezone from pathlib import Path from notify import ROOT, send BOX = "root@134.199.153.159" REMOTE_STATE = "/root/personal-cos/state" RECEIPTS = ROOT / "state" / "watch_box.jsonl" # The box brief fires 18:30 daily. 26 hours means one missed day raises it, not two. STALE_HOURS = 26 # Announce the fourteen-day mark once, so the definition of done closes itself instead # of relying on someone remembering to look. MILESTONE_DAYS = 14 def _receipt(verdict: str, detail: str) -> None: RECEIPTS.parent.mkdir(parents=True, exist_ok=True) with RECEIPTS.open("a", encoding="utf-8", newline="\n") as handle: handle.write( json.dumps( { "at": datetime.now(timezone.utc).isoformat(), "verdict": verdict, "detail": detail, } ) + "\n" ) def _already_announced(milestone: int) -> bool: """Has the milestone already been announced? Checks the verdict FIELD, not the line. This was a substring search over the raw line, and it matched a receipt whose *detail text* merely mentioned the milestone — so the real fourteenth day would have passed in silence because a note about a drill was sitting in the log. Grep over structured data reads whatever happens to be nearby. """ if not RECEIPTS.exists(): return False marker = f"milestone:{milestone}" for line in RECEIPTS.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue try: if json.loads(line).get("verdict") == marker: return True except ValueError: continue return False # Dates that must stay in the register until they pass. Not a general mechanism — # a short, deliberate list of the obligations whose silent disappearance would cost # the most. Add one when a fact becomes load-bearing; a passed date drops out on its # own, so the list does not need pruning. CANARIES = { "2026-09-23": "Skills Assessment - Carpenter (186 visa closes at 45)", } def _missing_canaries() -> list[tuple[str, str]]: """Which watched deadlines are gone from the box's register but not yet past? Returns [] on any read failure — an unreadable register is the UNREACHABLE path's problem, not a reason to cry that a deadline vanished. """ today = datetime.now().astimezone().date().isoformat() due_soon = {d: w for d, w in CANARIES.items() if d >= today} if not due_soon: return [] try: result = subprocess.run( [ "ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", BOX, "grep -ho '\"date\": \"[0-9-]*\"' /root/personal-cos/data/deadlines.jsonl " "| sort -u", ], capture_output=True, text=True, timeout=60, ) except (subprocess.TimeoutExpired, OSError): return [] if result.returncode != 0: return [] present = {line.split('"')[3] for line in result.stdout.splitlines() if '"' in line} return [(d, w) for d, w in sorted(due_soon.items()) if d not in present] def _read_remote(name: str, attempts: int = 3) -> dict | None: """Fetch one small JSON file off the box. Returns None only after real retries. ONE failed SSH is not an outage. Proven the hard way on 2026-07-29: this fired UNREACHABLE at Rocky while the box had an uptime of ten days and was answering pings — roughly thirty connections in quick succession had tripped the server's connection rate limiting, and the next attempt succeeded immediately. A watchdog that cries wolf on a dropped packet gets muted, and then it is muted on the night it is right. Three tries with a widening gap before saying anything. """ for attempt in range(attempts): try: result = subprocess.run( [ "ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", BOX, f"cat {REMOTE_STATE}/{name}", ], capture_output=True, text=True, timeout=90, ) if result.returncode == 0 and result.stdout.strip(): try: return json.loads(result.stdout) except ValueError: return None except (subprocess.TimeoutExpired, OSError): pass if attempt < attempts - 1: time.sleep(10 * (attempt + 1)) return None def main() -> int: try: last_brief = _read_remote("last_brief.json") except (subprocess.TimeoutExpired, OSError) as error: message = ( "PERSONAL RAIL — CANNOT REACH THE BOX\n" f"{type(error).__name__}: {error}\n" "The brief may not have run. Nothing on the box can tell you this." ) send(message) _receipt("UNREACHABLE", f"{type(error).__name__}: {error}") return 1 if last_brief is None: send( "PERSONAL RAIL — CANNOT REACH THE BOX\n" "SSH failed or state/last_brief.json is unreadable.\n" "The brief may not have run. Nothing on the box can tell you this." ) _receipt("UNREACHABLE", "ssh failed or last_brief.json unreadable") return 1 try: sent_at = datetime.fromisoformat(last_brief["sent_at_utc"]) except (KeyError, ValueError) as error: send(f"PERSONAL RAIL — box state is unreadable\n{type(error).__name__}: {error}") _receipt("BAD STATE", f"{type(error).__name__}: {error}") return 1 age = datetime.now(timezone.utc) - sent_at local = sent_at.astimezone().strftime("%a %d %b %H:%M") if age > timedelta(hours=STALE_HOURS): send( "PERSONAL RAIL — THE BOX HAS GONE QUIET\n" f"Last brief: {local} ({age.total_seconds() / 3600:.0f}h ago).\n" "The box is reachable, so its own heartbeat should have said this." ) _receipt("STALE", f"last brief {local}, {age.total_seconds() / 3600:.1f}h old") return 1 # The heartbeat proves a brief ARRIVED. It cannot prove the brief was RIGHT. # If the register lost its 23/09 entry — a bad rebuild, a truncated write, a # rules change that stopped matching — the brief would keep sending, the streak # would keep counting, and the one fact this system was built to carry would be # gone with nothing saying so. Check the fact itself, not just the delivery. missing = _missing_canaries() if missing: send( "PERSONAL RAIL — A WATCHED DEADLINE HAS VANISHED FROM THE REGISTER\n" + "\n".join(f"{due} {what}" for due, what in missing) + "\nThe brief is still sending. It is no longer carrying this." ) _receipt("CANARY LOST", "; ".join(f"{d} {w}" for d, w in missing)) return 1 streak = _read_remote("unattended.json") or {} days = int(streak.get("streak", 0)) if days >= MILESTONE_DAYS and not _already_announced(MILESTONE_DAYS): send( f"PERSONAL RAIL — {days} CONSECUTIVE UNATTENDED DAYS.\n" "The brief has landed every day without you touching it, from a machine " "that does not sleep. That was the definition of done." ) _receipt(f"milestone:{MILESTONE_DAYS}", f"streak reached {days}") return 0 _receipt("ok", f"box brief {local}, {age.total_seconds() / 3600:.1f}h old, streak {days}/14") return 0 if __name__ == "__main__": from error_reporting import record_failure try: raise SystemExit(main()) except SystemExit: raise except Exception as error: summary = record_failure("Box watcher failed", error) # A watcher that dies silently is the thing it exists to prevent. try: send(f"PERSONAL RAIL — the box watcher itself failed\n{summary}") except Exception: pass print(f"Box watcher failed: {summary}") raise SystemExit(1)