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, 62 insertions, 6 deletions

diff --git a/tools/guest-diag/semfix_watch.py b/tools/guest-diag/semfix_watch.py
index a4b2fc0..4b60fd6 100644
--- a/tools/guest-diag/semfix_watch.py
+++ b/tools/guest-diag/semfix_watch.py
@@ -58,12 +58,47 @@ CHAN_MAX_CANDS = int(os.environ.get("CHAN_MAX_CANDS", "256"))
 
 class Qmp:
     def __init__(self, path):
+        self.path = path
+        self._cache = {}
+        self._fails = 0
+        self._last_reconnect = 0.0
+        self._reconnecting = False
         self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
         self.s.connect(path)
         self.s.settimeout(15)
         self._rd()
         self.cmd("qmp_capabilities")
-        self._cache = {}
+
+    def reconnect(self):
+        """Re-establish the QMP connection after the socket died.
+
+        QEMU's QMP socket accepts a single client and closes when the guest
+        reboots; cmd() used to swallow every exception and return {}, so all
+        reads stayed empty forever ("uvmt(hb) sem=... unreadable") and
+        pinning stopped. Rate-limited to one attempt per second so a failing
+        monitor does not turn every read into a connect storm."""
+        if self._reconnecting or time.time() - self._last_reconnect < 1.0:
+            return
+        self._reconnecting = True
+        self._last_reconnect = time.time()
+        try:
+            try:
+                self.s.close()
+            except Exception:
+                pass
+            self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            self.s.connect(self.path)
+            self.s.settimeout(15)
+            self._rd()
+            self.cmd("qmp_capabilities")
+            self._cache = {}
+            self._ncpu = 0
+            self._cpu = 0
+            self._fails = 0
+        except Exception:
+            pass
+        finally:
+            self._reconnecting = False
 
     def _rd(self):
         """Read one QMP reply (JSON object ends with a newline). Poll with a
@@ -91,8 +126,17 @@ class Qmp:
     def cmd(self, name, **args):
         try:
             self.s.sendall(json.dumps({"execute": name, "arguments": args}).encode() + b"\n")
-            return json.loads(self._rd() or "{}")
+            reply = self._rd()
+            if not reply:
+                raise OSError("empty QMP reply")
+            self._fails = 0
+            return json.loads(reply)
         except Exception:
+            # Broken socket (QEMU restart, monitor hiccup): try exactly one
+            # reconnect, rate-limited inside reconnect().
+            self._fails = getattr(self, "_fails", 0) + 1
+            if not getattr(self, "_reconnecting", False):
+                self.reconnect()
             return {}
 
     def hmp(self, line):
@@ -143,8 +187,14 @@ class Qmp:
                 break
         m = re.search(r":(.*)$", out, re.S)
         if not m:
+            if not out.strip():
+                # Empty monitor output: the socket is broken, not the address.
+                self.reconnect()
             return []
         vals = [int(x, 16) for x in re.findall(r"0x([0-9a-fA-F]+)", m.group(1))][:n]
+        if not vals:
+            self.reconnect()
+            return []
         if cache:
             self._cache[key] = vals
         return vals
@@ -397,8 +447,11 @@ def resolve_pushbuffer_wide(q, sem, d):
 def pin(va, pa):
     cmd = "printf '0x%x,0x%x' > %s/sema_map_add" % (va, pa, PARAM)
     for _ in range(40):
-        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=5)
+        except subprocess.TimeoutExpired:
+            continue
         if r.returncode == 0 and "timed out" not in (r.stderr or ""):
             return True
         time.sleep(0.05)  # async queue full: drain and retry
@@ -408,8 +461,11 @@ def pin(va, pa):
 def pin_bulk(va, pa, nbytes):
     cmd = "printf '0x%x,0x%x,0x%x' > %s/sema_map_add" % (va, pa, nbytes, PARAM)
     for _ in range(40):
-        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=5)
+        except subprocess.TimeoutExpired:
+            continue
         if r.returncode == 0 and "timed out" not in (r.stderr or ""):
             return True
         time.sleep(0.05)