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.

4 files changed, 459 insertions, 77 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)
diff --git a/tools/guest-diag/pin_supervisor.sh b/tools/guest-diag/pin_supervisor.sh
new file mode 100755
index 0000000..5bbab0c
--- /dev/null
+++ b/tools/guest-diag/pin_supervisor.sh
@@ -0,0 +1,138 @@
+#!/bin/bash
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (c) 2026 Open-IOV contributors
+# pin_supervisor.sh - supervise the host-side pin tools (pin_gr_stream.py and
+# semfix_uvmt.py) so a silent death cannot strand every later CUDA client at
+# first cudaMalloc.
+#
+# Both tools write a heartbeat file; this script checks, every 5 s, that each
+# process is alive (exact script path) and that its heartbeat mtime is < 30 s.
+# On failure it kills the tool (exact path match) and restarts it, appending
+# RESTART lines to the supervisor log. After 10 restarts in one hour for a
+# tool it logs GIVEUP and exits 1. Only one semfix_uvmt instance is ever
+# started (QEMU's QMP socket accepts a single client).
+#
+# Usage: pin_supervisor.sh LOGDIR TAG
+# Logs: LOGDIR/12-pin-supervisor-TAG.log (supervisor)
+#       LOGDIR/12-pin_gr_stream-TAG.log, LOGDIR/12-semfix-TAG.log (tools)
+set -uo pipefail
+
+LOGDIR="${1:?usage: pin_supervisor.sh LOGDIR TAG}"
+TAG="${2:?usage: pin_supervisor.sh LOGDIR TAG}"
+
+HERE=$(cd "$(dirname "$0")" && pwd)
+PIN="$HERE/pin_gr_stream.py"
+SEMFIX="$HERE/semfix_uvmt.py"
+PIN_HB="${PIN_GR_HEARTBEAT:-/tmp/opencode/pin_gr_heartbeat}"
+SEMFIX_HB="${SEMFIX_HEARTBEAT_FILE:-/tmp/opencode/semfix_heartbeat}"
+SUPLOG="$LOGDIR/12-pin-supervisor-$TAG.log"
+PIN_LOG="$LOGDIR/12-pin_gr_stream-$TAG.log"
+SEMFIX_LOG="$LOGDIR/12-semfix-$TAG.log"
+WINDOW=3600
+CAP=10
+mkdir -p "$LOGDIR" /tmp/opencode
+
+log() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$SUPLOG"; }
+
+tool_alive() { pgrep -f "python3 $1" >/dev/null 2>&1; }
+
+hb_age() {  # age in seconds of the heartbeat file, 9999 if missing
+    local f="$1" m
+    m=$(stat -c %Y "$f" 2>/dev/null) || { echo 9999; return; }
+    echo $(( $(date +%s) - m ))
+}
+
+kill_tool() {
+    pkill -f "python3 $1" 2>/dev/null
+    sleep 1
+    if pgrep -f "python3 $1" >/dev/null 2>&1; then
+        pkill -9 -f "python3 $1" 2>/dev/null
+        sleep 1
+    fi
+    return 0
+}
+
+start_pin() {
+    if tool_alive "$PIN"; then return 0; fi
+    touch "$PIN_HB" 2>/dev/null || true
+    ( setsid nohup env SEMFIX_COUNTER_RETRY=10 \
+        python3 "$PIN" >> "$PIN_LOG" 2>&1 & )
+    log "START pin_gr_stream"
+}
+
+start_semfix() {
+    # pgrep exact-count check before start: if an instance already exists
+    # (ours or a leftover), never start a second one.
+    if tool_alive "$SEMFIX"; then
+        log "semfix already running - not starting another"
+        return 0
+    fi
+    touch "$SEMFIX_HB" 2>/dev/null || true
+    ( setsid nohup env SEMFIX_SKIP_PB=0 SEMFIX_COUNTER_RETRY=10 \
+        python3 "$SEMFIX" 0.5 >> "$SEMFIX_LOG" 2>&1 & )
+    log "START semfix_uvmt"
+}
+
+prune() {  # $1 now, $2 list -> list with entries older than WINDOW dropped
+    local now="$1" t out=""
+    for t in $2; do
+        [ $(( now - t )) -lt "$WINDOW" ] && out="$out $t"
+    done
+    printf '%s' "$out"
+}
+
+count_list() {  # $1 list -> element count
+    local n=0 t
+    for t in $1; do n=$(( n + 1 )); done
+    echo "$n"
+}
+
+restarts_pin=""
+restarts_semfix=""
+
+restart() {  # $1 label, $2 script path, $3 start function
+    local now list n
+    now=$(date +%s)
+    if [ "$1" = pin_gr_stream ]; then
+        restarts_pin=$(prune "$now" "$restarts_pin")
+        n=$(count_list "$restarts_pin")
+    else
+        restarts_semfix=$(prune "$now" "$restarts_semfix")
+        n=$(count_list "$restarts_semfix")
+    fi
+    if [ "$n" -ge "$CAP" ]; then
+        log "GIVEUP $1: $n restarts within ${WINDOW}s"
+        exit 1
+    fi
+    kill_tool "$2"
+    if [ "$1" = pin_gr_stream ]; then
+        restarts_pin="$restarts_pin $now"
+    else
+        restarts_semfix="$restarts_semfix $now"
+    fi
+    log "RESTART $(( n + 1 )) $1 $(date +%Y-%m-%dT%H:%M:%S)"
+    "$3"
+}
+
+check_tool() {  # $1 label, $2 path, $3 heartbeat file, $4 start function
+    local age
+    age=$(hb_age "$3")
+    if tool_alive "$2" && [ "$age" -le 30 ]; then
+        return 0
+    fi
+    if ! tool_alive "$2"; then
+        log "$1 not running (heartbeat age ${age}s) - restarting"
+    else
+        log "$1 heartbeat stale (${age}s) - restarting"
+    fi
+    restart "$1" "$2" "$4"
+}
+
+log "pin_supervisor up (LOGDIR=$LOGDIR TAG=$TAG)"
+start_pin
+start_semfix
+while true; do
+    sleep 5
+    check_tool pin_gr_stream "$PIN" "$PIN_HB" start_pin
+    check_tool semfix_uvmt "$SEMFIX" "$SEMFIX_HB" start_semfix
+done
diff --git a/tools/guest-diag/semfix_uvmt.py b/tools/guest-diag/semfix_uvmt.py
index cf54445..fa3ae01 100755
--- a/tools/guest-diag/semfix_uvmt.py
+++ b/tools/guest-diag/semfix_uvmt.py
@@ -53,6 +53,21 @@ SKIP_PB = os.environ.get("SEMFIX_SKIP_PB", "0") == "1"
 # Seconds between counter-page pin attempts (the guest-agent exec blocks for
 # its full timeout when the client saturates the guest).
 COUNTER_RETRY_S = float(os.environ.get("SEMFIX_COUNTER_RETRY", "10"))
+# Liveness file checked by pin_supervisor.sh (mtime must stay < 30 s).
+SEMFIX_HB_FILE = os.environ.get("SEMFIX_HEARTBEAT_FILE",
+                                "/tmp/opencode/semfix_heartbeat")
+# Every client pid ever observed. Module-level on purpose: the per-client
+# reset must NOT clear it, so a replayed ring entry for an old pid can never
+# masquerade as a new client and wipe the pin state.
+seen_pids = set()
+
+
+def _hb():
+    try:
+        with open(SEMFIX_HB_FILE, "w") as f:
+            f.write("%.3f\n" % time.time())
+    except OSError:
+        pass
 
 
 class Stream:
@@ -269,6 +284,17 @@ def main():
     counter_pids = set()
     counter_next_try = {}
     last_pid = None
+    pending_pid = None
+    pending_count = 0
+    # Stale-sem tracking: consecutive unreadable reads, first-unreadable time,
+    # and the per-sem read skip deadline once a sem has failed 3 times.
+    unread_count = {}
+    first_unreadable_t = {}
+    skip_until = {}
+    # Failed pool-pin tracking: retry with a 15 s backoff after 2 failures.
+    sem_pin_fail = {}
+    sem_pin_skip_until = {}
+    qmp_next_try = 0.0
     stream = Stream(SEMSTREAM)
     loop = 0
 
@@ -293,6 +319,7 @@ def main():
 
     while True:
         loop += 1
+        _hb()
         sems = set()
         pbs = set()
         chans = set()
@@ -302,17 +329,49 @@ def main():
             log("stream: new uvmt sem 0x%x" % s)
         for p in new_pids:
             log("stream: client pid %d" % p)
-            if last_pid is not None and p != last_pid:
-                stream.reset()
-                known_sems.clear()
-                seen_sem_pin.clear()
-                last_state.clear()
-                last_seen_t.clear()
-                pinned_ranges.clear()
-                seen_ch.clear()
-                seen_pb.clear()
-                counter_pids.clear()
-                log("semfix: new client pid %d - reset per-client pin state" % p)
+            if last_pid is None:
+                seen_pids.add(p)
+                last_pid = p
+                pending_pid = None
+                continue
+            if p == last_pid:
+                seen_pids.add(p)
+                continue
+            if p in seen_pids or p < last_pid:
+                # Replayed ring/stream entry for a client we already know, or
+                # a backwards transition: never reset on those.
+                log("semfix: pid %d seen before/backwards - no reset" % p)
+                seen_pids.add(p)
+                continue
+            if pending_pid != p:
+                pending_pid = p
+                pending_count = 1
+                log("semfix: pid change %d -> %d (debounce; reset if it "
+                    "repeats)" % (last_pid, p))
+                continue
+            pending_count += 1
+            if pending_count < 2:
+                continue
+            # Same unseen pid reported on two consecutive polls: real new
+            # client. seen_pids keeps old pids from ever triggering this.
+            seen_pids.add(p)
+            pending_pid = None
+            pending_count = 0
+            stream.reset()
+            known_sems.clear()
+            seen_sem_pin.clear()
+            last_state.clear()
+            last_seen_t.clear()
+            pinned_ranges.clear()
+            seen_ch.clear()
+            seen_pb.clear()
+            counter_pids.clear()
+            sem_pin_fail.clear()
+            sem_pin_skip_until.clear()
+            unread_count.clear()
+            first_unreadable_t.clear()
+            skip_until.clear()
+            log("semfix: new client pid %d - reset per-client pin state" % p)
             last_pid = p
         for pb in getattr(stream, "new_pbs", []):
             log("stream: new pushbuffer 0x%x" % pb)
@@ -327,6 +386,7 @@ def main():
                 t_sems, t_pbs, t_chans = guest_trace()
             except Exception:
                 t_sems, t_pbs, t_chans = set(), set(), set()
+            _hb()
             sems.update(t_sems)
             pbs.update(t_pbs)
             chans.update(t_chans)
@@ -337,12 +397,44 @@ def main():
             known_sems.add(sem)
 
         for sem in sorted(known_sems):
+            now = time.time()
+            if now < skip_until.get(sem, 0):
+                continue
             try:
                 st = uvmt_state(q, sem)
             except Exception:
                 st = None
+            if st is None:
+                # Stale-sem hygiene: the trace ring replays dead clients'
+                # kernel pointers forever; without eviction every loop burns
+                # QMP reads on them and the log floods with "unreadable".
+                cnt = unread_count.get(sem, 0) + 1
+                unread_count[sem] = cnt
+                first_unreadable_t.setdefault(sem, now)
+                if cnt >= 3:
+                    skip_until[sem] = now + 30.0
+                    if cnt == 3:
+                        log("semfix: sem=0x%x unreadable x3 - retrying every "
+                            "30s" % sem)
+                age = now - first_unreadable_t[sem]
+                if age >= 60.0 and sem not in stream.sems:
+                    known_sems.discard(sem)
+                    unread_count.pop(sem, None)
+                    first_unreadable_t.pop(sem, None)
+                    skip_until.pop(sem, None)
+                    last_state.pop(sem, None)
+                    last_seen_t.pop(sem, None)
+                    seen_sem_pin.discard(sem)
+                    sem_pin_fail.pop(sem, None)
+                    sem_pin_skip_until.pop(sem, None)
+                    log("semfix: evicting stale sem=0x%x (unreadable %.0fs, "
+                        "absent from stream)" % (sem, age))
+                    continue
+            else:
+                unread_count.pop(sem, None)
+                first_unreadable_t.pop(sem, None)
+                skip_until.pop(sem, None)
             key = state_key(st)
-            now = time.time()
             if key != last_state.get(sem):
                 last_state[sem] = key
                 last_seen_t[sem] = now
@@ -354,18 +446,33 @@ def main():
             # Pin the pool page once per (sem, word) resolution. The payload
             # word is what the releasing push writes; pinning the page makes
             # the host able to apply that release.
-            if st and sem not in seen_sem_pin:
+            if (st and sem not in seen_sem_pin
+                    and now >= sem_pin_skip_until.get(sem, 0)):
                 try:
                     pairs = resolve_sem_pool(q, sem)
                 except Exception as e:
                     pairs = []
                     log("semfix: sem=0x%x resolve failed: %s" % (sem, e))
                 if pairs:
-                    seen_sem_pin.add(sem)
+                    ok_any = False
                     for gv, gpa in pairs:
                         ok = pin(gv, gpa)
+                        ok_any = ok_any or ok
                         log("semfix: sem=0x%x pool va=0x%x -> gpa=0x%x pin=%s"
                             % (sem, gv, gpa, "ok" if ok else "FAILED"))
+                    if not ok_any:
+                        # Mark pinned only after a successful pin(): a failed
+                        # pin used to be recorded as done and never retried.
+                        fails = sem_pin_fail.get(sem, 0) + 1
+                        sem_pin_fail[sem] = fails
+                        if fails >= 2:
+                            sem_pin_skip_until[sem] = now + 15.0
+                            log("semfix: sem=0x%x pool pin failed x%d - retry "
+                                "in 15s" % (sem, fails))
+                        continue
+                    seen_sem_pin.add(sem)
+                    sem_pin_fail.pop(sem, None)
+                    sem_pin_skip_until.pop(sem, None)
                     if SKIP_PB:
                         continue
                     d = {}
@@ -393,6 +500,20 @@ def main():
                 else:
                     log("semfix: sem=0x%x pool unresolved (retry)" % sem)
 
+        if getattr(q, "_fails", 0) >= 5:
+            # QMP went away (QEMU restart, monitor died): all reads have been
+            # empty for 5 consecutive replies. Recreate the client loudly
+            # instead of reading empty forever ("uvmt(hb) ... unreadable").
+            now = time.time()
+            if now >= qmp_next_try:
+                log("semfix: QMP unreadable (%d consecutive empty replies) - "
+                    "recreating connection" % q._fails)
+                try:
+                    q = Qmp(QMP_SOCK)
+                except Exception as e:
+                    log("semfix: QMP recreate failed: %s" % e)
+                qmp_next_try = now + 5.0
+
         for ch in sorted(chans - seen_ch):
             d = {}
             try:
@@ -439,7 +560,9 @@ def main():
         # cadence: hammering the agent here starves it and wedges the run.
         if last_pid is None and loop % TRACE_EVERY == 0:
             last_pid = cdt_pid()
+            _hb()
             if last_pid:
+                seen_pids.add(last_pid)
                 log("counter: client pid %d (agent)" % last_pid)
         if last_pid and last_pid not in counter_pids:
             now = time.time()
@@ -448,6 +571,7 @@ def main():
                     pin_counter_page(last_pid, counter_pids)
                 except Exception as e:
                     log("counter: pin failed: %s" % e)
+                _hb()
                 counter_next_try[last_pid] = now + COUNTER_RETRY_S
 
         time.sleep(interval)
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)