layered-soul/skills/systematic-debugging/scripts/live_download_monitor.py

#!/usr/bin/env python3
"""Bounded network/download monitor for Wi-Fi/Tailscale/SSH investigations.

Writes compact JSONL samples. Designed to avoid Desktop clutter and avoid
unbounded packet captures during large downloads.

Environment variables:
  MON_INTERVAL       seconds between samples (default 10)
  MON_MAX_SECONDS    max runtime (default 1800)
  MON_MAX_BYTES      max log size (default 2 MiB)
  MON_WARN_FREE_GB   add warning below this free space (default 15)
  MON_STOP_FREE_GB   stop below this free space (default 10)
  MON_IFACE          Wi-Fi interface (default wlp1s0)
  MON_TARGETS        comma-separated label=ip:port TCP targets
  MON_PINGS          comma-separated ping targets
"""
import json, os, re, shutil, subprocess, sys, time
from datetime import datetime
from pathlib import Path

HOME = Path.home()
LOG_DIR = HOME / ".local/state/hermes/net-tests"
LOG_DIR.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
log_path = Path(sys.argv[1]) if len(sys.argv) > 1 else LOG_DIR / f"live-download-{stamp}.jsonl"
interval = int(os.environ.get("MON_INTERVAL", "10"))
max_seconds = int(os.environ.get("MON_MAX_SECONDS", "1800"))
max_bytes = int(os.environ.get("MON_MAX_BYTES", str(2 * 1024 * 1024)))
warn_free_gb = float(os.environ.get("MON_WARN_FREE_GB", "15"))
stop_free_gb = float(os.environ.get("MON_STOP_FREE_GB", "10"))
iface = os.environ.get("MON_IFACE", "wlp1s0")

def parse_targets(s):
    out = []
    for item in s.split(','):
        if not item.strip() or '=' not in item or ':' not in item:
            continue
        label, rest = item.split('=', 1)
        ip, port = rest.rsplit(':', 1)
        out.append((label.strip(), ip.strip(), port.strip()))
    return out

TARGETS = parse_targets(os.environ.get(
    "MON_TARGETS",
    "osa_public_https=51.83.197.148:443,osa_tailscale_ssh=100.72.229.63:22,domedog_tailscale_ssh=100.103.255.41:22",
))
PING_TARGETS = [x.strip() for x in os.environ.get("MON_PINGS", "10.91.179.29,1.1.1.1,100.72.229.63,100.103.255.41").split(',') if x.strip()]

def run(cmd, timeout=8):
    try:
        return subprocess.run(cmd, shell=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout).stdout
    except Exception as e:
        return f"ERR: {e}"

def ping_summary(target):
    out = run(f"ping -c 3 -i 0.2 -W 2 {target}", timeout=5)
    rec = {"target": target, "loss": None, "avg_ms": None, "max_ms": None, "mdev_ms": None}
    m = re.search(r"(\d+) packets transmitted, (\d+) received,\s*([0-9.]+)% packet loss", out)
    if m: rec["loss"] = float(m.group(3))
    r = re.search(r"rtt min/avg/max/mdev = ([0-9.]+)/([0-9.]+)/([0-9.]+)/([0-9.]+)", out)
    if r:
        rec.update({"avg_ms": float(r.group(2)), "max_ms": float(r.group(3)), "mdev_ms": float(r.group(4))})
    return rec

def parse_ss():
    lines = run("ss -tinp", timeout=5).splitlines()
    conns, current = [], None
    for line in lines:
        if line.startswith(("ESTAB", "FIN-", "CLOSE-WAIT", "SYN-")):
            parts = line.split()
            if len(parts) >= 5:
                current = {"state": parts[0], "recvq": parts[1], "sendq": parts[2], "local": parts[3], "peer": parts[4]}
                conns.append(current)
        elif current and "cubic" in line:
            m = re.search(r"rtt:([0-9.]+)/([0-9.]+)", line)
            if m:
                current["rtt_ms"] = float(m.group(1)); current["rtt_var_ms"] = float(m.group(2))
            for key, pat in {
                "bytes_received": r"bytes_received:(\d+)", "bytes_sent": r"bytes_sent:(\d+)",
                "bytes_retrans": r"bytes_retrans:(\d+)", "retrans_total": r"retrans:\d+/(\d+)",
                "reord_seen": r"reord_seen:(\d+)", "rcv_ooopack": r"rcv_ooopack:(\d+)", "cwnd": r"cwnd:(\d+)",
            }.items():
                mm = re.search(pat, line)
                if mm: current[key] = int(mm.group(1))
    selected = []
    for c in conns:
        for label, ip, port in TARGETS:
            if ip in c.get("peer", "") and c.get("peer", "").endswith(":" + port):
                selected.append({**c, "label": label})
    return selected

def disk_state():
    u = shutil.disk_usage(str(HOME))
    return {"total_gb": round(u.total/1e9,2), "used_gb": round(u.used/1e9,2), "free_gb": round(u.free/1e9,2), "free_pct": round(u.free/u.total*100,2)}

def wifi_state():
    cmd = f"nmcli -t -f GENERAL.CONNECTION,IP4.GATEWAY device show {iface} 2>/dev/null; nmcli -f IN-USE,SSID,BSSID,CHAN,FREQ,RATE,SIGNAL dev wifi list --rescan no 2>/dev/null | sed -n '1,5p'"
    return run(cmd, timeout=5).strip().splitlines()[:8]

start = time.time()
with log_path.open("a", buffering=1) as f:
    f.write(json.dumps({"type":"start", "ts": datetime.now().isoformat(), "interval_s": interval, "max_seconds": max_seconds, "max_bytes": max_bytes, "note":"bounded live monitor; no packet payload capture"}) + "\n")
    while True:
        ds = disk_state()
        rec = {"type":"sample", "ts": datetime.now().isoformat(timespec="seconds"), "elapsed_s": round(time.time()-start,1), "disk": ds, "wifi": wifi_state(), "tcp": parse_ss(), "ping": [ping_summary(t) for t in PING_TARGETS]}
        if ds["free_gb"] <= warn_free_gb: rec["warning"] = f"low disk free {ds['free_gb']}GB"
        f.write(json.dumps(rec, separators=(",", ":")) + "\n")
        if ds["free_gb"] <= stop_free_gb: reason = "disk safety threshold"
        elif time.time() - start >= max_seconds: reason = "max_seconds reached"
        elif log_path.stat().st_size >= max_bytes: reason = "max log size reached"
        else: reason = None
        if reason:
            f.write(json.dumps({"type":"stop", "ts": datetime.now().isoformat(), "reason": reason, "disk": ds, "size": log_path.stat().st_size}) + "\n")
            break
        time.sleep(interval)
print(log_path)