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, 138 insertions, 14 deletions

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)