vgsp: m177 pin-free self-sufficiency wave + tool hardening

(1) svc_session_flush keeps svc.direct_base (per-boot constant; the pin-free
    targeted scanner's gate/input) and a new gsp_sema_service_guest_reset()
    hook wipes boot-scoped discovery state on guest RM reload;
(2) counter-page derive now takes the first READABLE source (vaspaces ->
    discovered -> targeted) instead of failing -EIO on a walked-but-garbage
    page, with one-shot diagnostics;
(3) new m177c UVM ring-space directory discovery: content-verified
    (push_va -> push_pa) samples from the GR content discovery and CE push
    pairing feed a second discovered-directory slot (walks the 0x2xxxxxxx
    family), consulted by svc_walk_discovered;
(4) GR channels now deliver withheld opdef sets at entry consume (normal
    tail and the stale-skip path) instead of waiting for the 10 s sweep;
(5) host tools hardened: semfix_uvmt QMP reconnect + pid-generation dedupe
    + stale-sem eviction + post-success pin latch + heartbeats; pin_gr_stream
    accept timeout + guest-scanner reinstall backoff + restart loop +
    subprocess timeouts; new pin_supervisor.sh (heartbeat watchdog, capped
    restarts) wired into start_vm_keep.sh.
Evidence: r194-r196 cycles; pool resolves src=trusted pin-free; ring dir
0x10801c000 found 4/4; row: 'rel-apply ... written' with tools up.

1 files changed, 121 insertions, 57 deletions

diff --git a/tools/guest-diag/pin_gr_stream.py b/tools/guest-diag/pin_gr_stream.py
index dff2d2d..0072b20 100644
--- a/tools/guest-diag/pin_gr_stream.py
+++ b/tools/guest-diag/pin_gr_stream.py
@@ -32,12 +32,22 @@ LO = os.environ.get("PINWIN_LO", "0x200000000")
 HI = os.environ.get("PINWIN_HI", "0x206000000")
 PAMAX = os.environ.get("PINWIN_PA_MAX", "0x180000000")
 SEMSTREAM = os.environ.get("SEM_STREAM_FILE", "/tmp/opencode/sem_stream.txt")
+# Liveness file checked by pin_supervisor.sh (mtime must stay < 30 s).
+HEARTBEAT_FILE = os.environ.get("PIN_GR_HEARTBEAT", "/tmp/opencode/pin_gr_heartbeat")
 
 
 def log(msg):
     print("[%s] %s" % (time.strftime("%H:%M:%S"), msg), flush=True)
 
 
+def _heartbeat():
+    try:
+        with open(HEARTBEAT_FILE, "w") as f:
+            f.write("%.3f\n" % time.time())
+    except OSError:
+        pass
+
+
 def _mirror(line):
     """Append guest control lines (#pid/#sem) to the shared host file so the
     single-QMP watcher can consume them without guest-agent round trips."""
@@ -60,7 +70,8 @@ def vexec(cmd, timeout=40):
 def _start_cmd():
     with open(GUEST_SCRIPT) as f:
         b64 = subprocess.run(["base64", "-w0"], stdin=f,
-                             capture_output=True, text=True).stdout
+                             capture_output=True, text=True,
+                             timeout=10).stdout
     prefixes = os.environ.get("CLIENT_PREFIXES", "cdt,va,victory")
     vma_every = os.environ.get("STREAM_VMA_EVERY", "1")
     # Kill EVERY existing scanner, not just the pid-file one: a retry after
@@ -90,6 +101,7 @@ def install_and_start():
     # or a stale qga client); retry a few times instead of giving up, because
     # a boot without the scanner streams no pins and the client then hangs.
     for attempt in range(5):
+        _heartbeat()
         out = vexec(_start_cmd(), 60)
         if any(ch.isdigit() for ch in out):
             return True
@@ -101,8 +113,12 @@ def install_and_start():
 def inject(va, pa, ln):
     cmd = "printf '0x%x,0x%x,0x%x,1' > %s/sema_map_add" % (va, pa, ln, PARAM)
     for _ in range(60):
-        r = subprocess.run(["sudo", "-n", "sh", "-c", cmd],
-                           capture_output=True, text=True)
+        try:
+            r = subprocess.run(["sudo", "-n", "sh", "-c", cmd],
+                               capture_output=True, text=True, timeout=10)
+        except subprocess.TimeoutExpired:
+            time.sleep(0.05)
+            continue
         if r.returncode == 0:
             return True
         time.sleep(0.05)
@@ -115,8 +131,12 @@ def scrub_register(fin_pa, hsem_pa):
     cmd = ("printf '0x%x,0x%x' > %s/sema_scrubbuf_add"
            % (fin_pa, hsem_pa, PARAM))
     for _ in range(60):
-        r = subprocess.run(["sudo", "-n", "sh", "-c", cmd],
-                           capture_output=True, text=True)
+        try:
+            r = subprocess.run(["sudo", "-n", "sh", "-c", cmd],
+                               capture_output=True, text=True, timeout=10)
+        except subprocess.TimeoutExpired:
+            time.sleep(0.05)
+            continue
         if r.returncode == 0:
             log("scrub registered fin=0x%x hsem=0x%x" % (fin_pa, hsem_pa))
             return True
@@ -166,81 +186,116 @@ def _demand_loop():
         log("demand: sent %d stalled VA(s) to guest" % len(todo))
 
 
+_DEMAND_STARTED = False
+
+
 def main():
+    global _DEMAND_STARTED
+    _heartbeat()
     if not install_and_start():
         log("WARN: could not start guest scanner (will retry via vsock wait)")
-    try:
-        threading.Thread(target=_demand_loop, daemon=True).start()
-    except Exception as e:
-        log("WARN: demand thread failed to start: %s" % e)
+    if not _DEMAND_STARTED:
+        _DEMAND_STARTED = True
+        try:
+            threading.Thread(target=_demand_loop, daemon=True).start()
+        except Exception as e:
+            log("WARN: demand thread failed to start: %s" % e)
     srv = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
     srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     srv.bind((socket.VMADDR_CID_ANY, PORT))
+    srv.settimeout(10)
     srv.listen(4)
     log("listening on vsock port %d (window %s..%s)" % (PORT, LO, HI))
 
     seen = {}
     pin_total = 0
-    buf = b""
+    last_conn = time.time()
+    last_reinstall = 0.0
     while True:
         try:
             conn, _ = srv.accept()
+        except socket.timeout:
+            now = time.time()
+            # No guest scanner connection for 30 s: it died or never started
+            # (a boot without it streams no pins and every client then hangs).
+            # Re-run install_and_start() at most once per 15 s.
+            if now - last_conn > 30.0 and now - last_reinstall >= 15.0:
+                last_reinstall = now
+                log("reinstall attempt: no guest connection for %.0fs"
+                    % (now - last_conn))
+                try:
+                    install_and_start()
+                except Exception as e:
+                    log("reinstall failed: %s" % e)
+            continue
         except OSError as e:
             log("accept failed: %s" % e)
             time.sleep(1)
             continue
+        last_conn = time.time()
+        _heartbeat()
         conn.settimeout(5)
         log("guest connected")
         buf = b""
         runs = []
-        while True:
-            try:
-                chunk = conn.recv(1 << 20)
-            except socket.timeout:
-                if runs:
-                    _flush(runs, seen)
-                    runs = []
-                continue
-            except OSError:
-                break
-            if not chunk:
-                break
-            buf += chunk
-            while b"\n" in buf:
-                line, buf = buf.split(b"\n", 1)
-                text = line.decode(errors="replace").strip()
-                if not text:
-                    continue
-                if text.startswith("#pid"):
-                    m = re.search(r"(\d+)", text)
-                    seen = {}
-                    log("guest reports cdt pid %s" % (m.group(1) if m else "?"))
-                    _mirror(text)
-                    continue
-                if text.startswith("#sem"):
-                    _mirror(text)
-                    continue
-                if text.startswith("#pb"):
-                    _mirror(text)
-                    continue
-                if text.startswith("#put"):
-                    log(text)
-                    _mirror(text)
+        try:
+            while True:
+                try:
+                    chunk = conn.recv(1 << 20)
+                except socket.timeout:
+                    if runs:
+                        _flush(runs, seen)
+                        runs = []
                     continue
-                if text.startswith("#scrub"):
-                    m = re.search(r"0x([0-9a-fA-F]+),0x([0-9a-fA-F]+)", text)
-                    _mirror(text)
+                except OSError:
+                    break
+                if not chunk:
+                    break
+                _heartbeat()
+                buf += chunk
+                while b"\n" in buf:
+                    line, buf = buf.split(b"\n", 1)
+                    text = line.decode(errors="replace").strip()
+                    if not text:
+                        continue
+                    if text.startswith("#pid"):
+                        m = re.search(r"(\d+)", text)
+                        seen = {}
+                        log("guest reports cdt pid %s"
+                            % (m.group(1) if m else "?"))
+                        _mirror(text)
+                        continue
+                    if text.startswith("#sem"):
+                        _mirror(text)
+                        continue
+                    if text.startswith("#pb"):
+                        _mirror(text)
+                        continue
+                    if text.startswith("#put"):
+                        log(text)
+                        _mirror(text)
+                        continue
+                    if text.startswith("#scrub"):
+                        m = re.search(r"0x([0-9a-fA-F]+),0x([0-9a-fA-F]+)",
+                                      text)
+                        _mirror(text)
+                        if m:
+                            scrub_register(int(m.group(1), 16),
+                                           int(m.group(2), 16))
+                        continue
+                    m = re.match(r"^(0x[0-9a-f]+),(0x[0-9a-f]+)(?:,(\d+))?$",
+                                 text)
                     if m:
-                        scrub_register(int(m.group(1), 16),
-                                       int(m.group(2), 16))
-                    continue
-                m = re.match(r"^(0x[0-9a-f]+),(0x[0-9a-f]+)(?:,(\d+))?$", text)
-                if m:
-                    runs.append((int(m.group(1), 16), int(m.group(2), 16),
-                                 int(m.group(3)) if m.group(3) else 1))
-            if runs:
-                pin_total += _flush(runs, seen)
-                runs = []
+                        runs.append((int(m.group(1), 16),
+                                     int(m.group(2), 16),
+                                     int(m.group(3)) if m.group(3) else 1))
+                if runs:
+                    pin_total += _flush(runs, seen)
+                    runs = []
+        except Exception as e:
+            # A transient error on one connection must not kill the process:
+            # log it, drop the connection and go back to accept().
+            log("connection handler error: %s (continuing)" % e)
         log("guest disconnected (pins this session: %d)" % pin_total)
         try:
             conn.close()
@@ -289,6 +344,7 @@ def _flush(runs, seen):
     CHUNK = 200
     for start in range(0, len(script), CHUNK):
         batch = script[start:start + CHUNK]
+        _heartbeat()
         try:
             r = subprocess.run(["sudo", "-n", "sh", "-c", "\n".join(batch)],
                                capture_output=True, text=True, timeout=60)
@@ -322,4 +378,12 @@ def _flush(runs, seen):
 
 
 if __name__ == "__main__":
-    main()
+    # main() loops forever; only an unexpected exception exits it. Restart
+    # instead of dying silently - a dead listener means no pins and every
+    # later CUDA client hangs at first cudaMalloc.
+    while True:
+        try:
+            main()
+        except Exception as e:
+            log("FATAL: main() exited: %r - restarting in 2s" % (e,))
+            time.sleep(2)