tools: kill-cascade purge + stall watchdogs + monotonic client identity (tools-fix c8fb5377)

Dead-pid purge via in-band #nopid (zombie-aware probe), monotonic
(pid,starttime) client identity so trace-ring replays can never reset an
active client (fixes the m178 mid-run reset regression avenue), pre-capture
and post-capture stall watchdogs that re-assert pins and escalate loudly,
heartbeat threads for both tools, supervisor STALE 75s/GRACE 120s.

1 files changed, 415 insertions, 72 deletions

diff --git a/tools/guest-diag/semfix_uvmt.py b/tools/guest-diag/semfix_uvmt.py
index fa3ae01..f28efd3 100755
--- a/tools/guest-diag/semfix_uvmt.py
+++ b/tools/guest-diag/semfix_uvmt.py
@@ -53,13 +53,24 @@ 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).
+# Liveness file checked by pin_supervisor.sh (mtime must stay < 75 s; the
+# supervisor grants a 120 s startup grace on top).
 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()
+# Dead-pid/stall watchdog tuning (seconds). The tracked client is probed in
+# the guest only when it has been idle this long, on a cadence, so live
+# clients are never probed and a killed client is detected within ~1 min.
+LIVENESS_IDLE_S = float(os.environ.get("SEMFIX_LIVENESS_IDLE", "15"))
+LIVENESS_EVERY_S = float(os.environ.get("SEMFIX_LIVENESS_EVERY", "25"))
+# Pre-capture stall: client alive, counter page not pinned, no progress for
+# this long -> recovery attempt; escalate loudly if it persists.
+STALL_PRECAP_S = float(os.environ.get("SEMFIX_STALL_PRECAP", "60"))
+STALL_RETRY_S = float(os.environ.get("SEMFIX_STALL_RETRY", "30"))
+STALL_ESCALATE_S = float(os.environ.get("SEMFIX_STALL_ESCALATE", "180"))
+# Inter-tool re-assert request: written here, consumed by pin_gr_stream.py
+# (full ping re-assert of every pair it has ever pinned).
+REASSERT_REQ = os.environ.get("PIN_REASSERT_FILE",
+                              "/tmp/opencode/pin_reassert_request")
 
 
 def _hb():
@@ -73,9 +84,16 @@ def _hb():
 class Stream:
     """Tail the guest scanner's control lines mirrored by pin_gr_stream.py.
 
-    The guest emits "#pid <n>" and "#sem 0x<k>" over vsock; the listener
-    appends them here. This replaces per-loop guest-agent trace reads, which
-    were too slow to catch short-lived tracking semaphores.
+    The guest emits "#pid <n> [starttime]" and "#sem 0x<k>" over vsock; the
+    listener appends them here. This replaces per-loop guest-agent trace
+    reads, which were too slow to catch short-lived tracking semaphores.
+
+    Replay safety (the m178 regression): the file persists across tool
+    restarts, so a fresh process re-reads the whole history. Lines written
+    before this process started (byte offset < init_size) are absorbed as
+    HISTORY - they seed the dedupe/identity state and the sem/pb lists but
+    can never trigger a per-client reset. Only lines appended afterwards
+    are live. This makes a supervisor restart mid-run harmless.
     """
 
     def __init__(self, path):
@@ -85,6 +103,15 @@ class Stream:
         self.pids = []
         self.pbs = []
         self.new_pbs = []
+        self.new_nopids = []
+        self.history_pids = []      # (pid, start) identities from history
+        self.pid_keys = set()       # (pid, start) dedupe, per generation
+        try:
+            # Everything already in the file when this process starts is
+            # history (a restart replays it); never reset on it.
+            self._init_size = os.path.getsize(path)
+        except OSError:
+            self._init_size = 0
 
     def reset(self):
         """Forget everything learned about the previous client. A new client
@@ -95,6 +122,16 @@ class Stream:
         self.pbs = []
         self.pids = []
         self.new_pbs = []
+        self.new_nopids = []
+        self.pid_keys = set()
+
+    @staticmethod
+    def _ident(line):
+        """Parse '#pid 1234' or '#pid 1234 5678900' -> (pid, start|None)."""
+        nums = re.findall(r"\d+", line)
+        if not nums:
+            return None, None
+        return int(nums[0]), (int(nums[1]) if len(nums) > 1 else None)
 
     def poll(self):
         try:
@@ -105,12 +142,19 @@ class Stream:
                     self.off = 0
                 f.seek(self.off)
                 data = f.read()
+                base = self.off
                 self.off = f.tell()
         except OSError:
-            return [], []
+            return [], [], []
         new_sems, new_pids = [], []
         self.new_pbs = []
+        self.new_nopids = []
+        pos = base
         for line in data.splitlines():
+            # A line is live when it ends past the startup boundary; history
+            # lines are absorbed (identity/dedupe) but never emitted as live.
+            live = pos + len(line) + 1 > self._init_size
+            pos += len(line) + 1
             if line.startswith("#sem"):
                 m = re.search(r"0x([0-9a-fA-F]+)", line)
                 if m:
@@ -126,13 +170,116 @@ class Stream:
                         self.pbs.append(v)
                         self.new_pbs.append(v)
             elif line.startswith("#pid"):
-                m = re.search(r"(\d+)", line)
-                if m:
-                    v = int(m.group(1))
-                    if v not in self.pids:
-                        self.pids.append(v)
-                        new_pids.append(v)
-        return new_sems, new_pids
+                pid, start = self._ident(line)
+                if pid is not None and (pid, start) not in self.pid_keys:
+                    self.pid_keys.add((pid, start))
+                    self.pids.append(pid)
+                    if live:
+                        new_pids.append((pid, start))
+                    else:
+                        self.history_pids.append((pid, start))
+            elif line.startswith("#nopid"):
+                # In-band client-death signal (new guest scanner; older ones
+                # simply never send it and the host probe covers the case).
+                pid, start = self._ident(line)
+                if live and pid is not None:
+                    self.new_nopids.append((pid, start))
+        return new_sems, new_pids, self.new_nopids
+
+
+class ClientTracker:
+    """Monotonic, replay-proof client-identity tracking.
+
+    A client's identity is (pid, starttime). Rules, in priority order:
+      * an identity adopted once is never reset a second time (duplicate
+        lines, trace/stream replays, supervisor restarts);
+      * a backwards pid, or the same pid with an older starttime, is a
+        replay - never reset;
+      * a pid recycled with a NEWER starttime is a real new client - reset
+        immediately (the previous holder is definitionally gone);
+      * a newer unseen identity only resets the per-client state once the
+        CURRENT client is CONFIRMED gone - a reset fired while the previous
+        client was still active is the m178 third-client hang regression.
+
+    Pure logic: the caller feeds guest liveness results in and performs the
+    actual state purge when an action says so.
+    """
+
+    def __init__(self):
+        self.seen = set()
+        self.current = None
+        self.pending = None
+        self.dead = set()
+
+    @staticmethod
+    def _start_match(a, b):
+        # Unknown starttimes match conservatively (legacy scanner lines).
+        return a[1] == b[1] or a[1] is None or b[1] is None
+
+    def observe_history(self, pid, start):
+        """Absorb an identity found in pre-restart history: the process has
+        no per-client state yet, so history only seeds identity tracking and
+        the freshest identity becomes current - no reset is ever requested
+        for history lines."""
+        key = (pid, start)
+        if key in self.seen:
+            return
+        self.seen.add(key)
+        self.current = key
+
+    def note_dead(self, pid, start):
+        """A #nopid or a successful dead probe: identity is gone."""
+        key = (pid, start)
+        self.dead.add(key)
+        if self.current and self.current[0] == pid \
+                and self._start_match(self.current, key):
+            self.current = None
+        if self.pending and self.pending[0] == pid \
+                and self._start_match(self.pending, key):
+            self.pending = None
+
+    def is_current(self, pid, start):
+        """True when (pid, start) is the currently tracked identity (used to
+        decide whether a death signal may purge - a stale #nopid for a
+        recycled pid must never purge the new holder)."""
+        return self.current is not None and self.current[0] == pid \
+            and self._start_match(self.current, (pid, start))
+
+    def decide(self, pid, start):
+        """Classify a live #pid line: adopt | skip | skip_backwards |
+        maybe_reset | reset."""
+        key = (pid, start)
+        if key in self.seen:
+            return "skip"
+        self.seen.add(key)
+        cur = self.current
+        if cur is None:
+            self.current = key
+            return "adopt"
+        if pid == cur[0] and self._start_match(cur, key):
+            return "skip"
+        if pid == cur[0] and start is not None and cur[1] is not None \
+                and start > cur[1]:
+            # Recycled pid, newer process: safe to reset now.
+            self.current = key
+            return "reset"
+        if pid < cur[0] or (pid == cur[0] and start is not None
+                            and cur[1] is not None and start < cur[1]):
+            return "skip_backwards"
+        self.pending = key
+        return "maybe_reset"
+
+    def resolve(self, old_dead):
+        """Caller learned whether the CURRENT client is dead. Returns
+        'reset' (purge + adopt the pending client) or 'defer'."""
+        if self.pending is None:
+            return None
+        if old_dead:
+            key = self.pending
+            self.pending = None
+            self.current = key
+            return "reset"
+        return "defer"
 
 
 def log(msg):
@@ -211,7 +358,9 @@ def dmesg_releases(seen):
 
     Only lines after the last "sema-svc: thread started" marker count: dmesg
     keeps the previous module load's lines and the old client's releases.
+    Returns the number of newly printed lines (used as watchdog progress).
     """
+    n = 0
     try:
         r = subprocess.run(
             ["sudo", "-n", "sh", "-c",
@@ -220,13 +369,15 @@ def dmesg_releases(seen):
              "| grep 'sema-push' | tail -300"],
             capture_output=True, text=True, timeout=10)
     except Exception:
-        return
+        return 0
     for line in r.stdout.splitlines():
         line = line.strip()
         if not line or line in seen:
             continue
         seen.add(line)
+        n += 1
         log("push: %s" % line.replace("vgsp: ", ""))
+    return n
 
 
 def cdt_pid():
@@ -239,6 +390,34 @@ def cdt_pid():
     return int(pids[-1]) if pids else None
 
 
+def _pid_alive_cmd(pid):
+    """Guest shell that reports ALIVE / DEAD for a pid. Zombie-aware: a
+    killed (SIGKILLed) client lingers as a defunct /proc entry, and a plain
+    `test -d /proc/<pid>` would latch it as alive forever (the m178
+    dead-pid latch)."""
+    return ("f=/proc/%d/stat; if [ -r \"$f\" ]; then "
+            "s=$(sed -e 's/^[^)]*) //' -e 's/ .*//' \"$f\"); "
+            "if [ \"$s\" = Z ] || [ -z \"$s\" ]; then echo DEAD; "
+            "else echo ALIVE; fi; else echo DEAD; fi" % pid)
+
+
+def pid_alive(pid):
+    """Guest liveness of a client pid: True alive, False dead/zombie/gone,
+    None unknown (agent error). Callers must treat None as alive (never
+    purge a client on an inconclusive probe)."""
+    try:
+        r = subprocess.run([VMEXEC, "exec", _pid_alive_cmd(pid)],
+                           capture_output=True, text=True, timeout=20)
+    except Exception:
+        return None
+    out = r.stdout or ""
+    if "ALIVE" in out:
+        return True
+    if "DEAD" in out:
+        return False
+    return None
+
+
 def pin_counter_page(pid, pinned_pids):
     """Pin the client's completion-counter page from its own pagemap.
 
@@ -283,9 +462,10 @@ def main():
     dseen = set()
     counter_pids = set()
     counter_next_try = {}
+    # Current client pid/starttime, mirrored from the tracker (starttime
+    # disambiguates recycled pids).
     last_pid = None
-    pending_pid = None
-    pending_count = 0
+    last_start = None
     # 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 = {}
@@ -296,8 +476,73 @@ def main():
     sem_pin_skip_until = {}
     qmp_next_try = 0.0
     stream = Stream(SEMSTREAM)
+    tracker = ClientTracker()
+    # Dead-pid/backwards-transition proofing and the stall watchdog.
+    history_absorbed = False
+    adopted_from_history = False
+    probe_next = 0.0            # next allowed idle-liveness probe
+    recover_next = 0.0          # next watchdog recovery attempt
+    stall_since = None          # first observation of the pre-capture stall
+    escalated = False
+    progress_t = time.time()    # last host-side progress for the client
     loop = 0
 
+    def _clear_client_state():
+        """Purge every per-client latch. ONLY valid when the previous client
+        is CONFIRMED dead (or at fresh process start): the m178 third-client
+        hang came from wiping the state of an ACTIVE client on a replayed
+        (#pid) identity."""
+        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()
+        counter_next_try.clear()
+        sem_pin_fail.clear()
+        sem_pin_skip_until.clear()
+        unread_count.clear()
+        first_unreadable_t.clear()
+        skip_until.clear()
+
+    def _adopt(pid, start, why):
+        nonlocal last_pid, last_start, progress_t, adopted_from_history
+        nonlocal stall_since, escalated
+        last_pid, last_start = pid, start
+        progress_t = time.time()
+        adopted_from_history = False
+        stall_since = None
+        escalated = False
+        log("semfix: tracking client pid %d%s (%s)"
+            % (pid, " start=%s" % start if start is not None else "", why))
+
+    def _watchdog_reset():
+        nonlocal stall_since, escalated
+        stall_since = None
+        escalated = False
+
+    def _watchdog_recover():
+        """Idempotent pre-capture stall recovery: re-arm every sem latch so
+        the next loop re-resolves and re-pins, force an immediate counter
+        retry, and request a full pin re-assert from pin_gr_stream."""
+        for sem in set(known_sems) | set(stream.sems):
+            skip_until.pop(sem, None)
+            sem_pin_skip_until.pop(sem, None)
+            sem_pin_fail.pop(sem, None)
+            unread_count.pop(sem, None)
+            first_unreadable_t.pop(sem, None)
+            seen_sem_pin.discard(sem)
+        if last_pid is not None:
+            counter_next_try[last_pid] = 0.0
+        try:
+            with open(REASSERT_REQ, "w") as f:
+                f.write("%.3f\n" % time.time())
+        except OSError:
+            pass
+
     def pin_range_once(gpu_vas, cpu, gpa, size=0):
         key = tuple(gpu_vas), cpu, size
         st = pinned_ranges.get(key)
@@ -324,64 +569,94 @@ def main():
         pbs = set()
         chans = set()
 
-        new_sems, new_pids = stream.poll()
+        new_sems, new_pids, new_nopids = stream.poll()
         for s in new_sems:
             log("stream: new uvmt sem 0x%x" % s)
-        for p in new_pids:
-            log("stream: client pid %d" % 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
+        if new_sems and last_pid is not None:
+            progress_t = time.time()
+
+        # In-band client-death signals (new guest scanner): purge the dead
+        # client's per-client state immediately so the NEXT client works
+        # without a VM cycle (the kill cascade).
+        for p, st in new_nopids:
+            log("stream: client gone (%d%s)"
+                % (p, " start=%s" % st if st is not None else ""))
+            was_current = tracker.is_current(p, st)
+            tracker.note_dead(p, st)
+            if was_current and last_pid == p:
+                _clear_client_state()
+                last_pid = last_start = None
+                _watchdog_reset()
+                log("semfix: purged per-client state of exited pid %d" % p)
+        # The old client is gone: adopt any client whose reset was deferred
+        # while it was still (assumed) alive.
+        if last_pid is None and tracker.pending is not None \
+                and tracker.current is None:
+            if tracker.resolve(old_dead=True) == "reset":
+                p, st = tracker.current
+                _clear_client_state()
+                _adopt(p, st, "deferred reset")
+        # Pre-restart history (a supervisor restart re-reads the file): absorb
+        # identities WITHOUT any reset - the fresh process already has empty
+        # state; the freshest identity is the client that was running.
+        if stream.history_pids and not history_absorbed:
+            for p, st in stream.history_pids:
+                tracker.observe_history(p, st)
+            history_absorbed = True
+            if tracker.current is not None and last_pid is None:
+                p, st = tracker.current
+                _adopt(p, st, "resumed from pre-restart history")
+                adopted_from_history = True
+
+        for p, st in new_pids:
+            cur_before = tracker.current
+            action = tracker.decide(p, st)
+            tag = " start=%s" % st if st is not None else ""
+            if action == "skip":
+                log("stream: client pid %d%s seen before - no reset" % (p, tag))
+            elif action == "skip_backwards":
+                log("semfix: client pid %d%s backwards/replay - no reset"
+                    % (p, tag))
+            elif action == "adopt":
+                log("stream: client pid %d%s" % (p, tag))
+                _adopt(p, st, "new client")
+            elif action == "reset":
+                # Recycled pid, newer starttime: the previous holder of this
+                # pid is definitionally gone, safe to reset now.
+                log("semfix: pid %d recycled (start=%s) - reset per-client "
+                    "pin state" % (p, st))
+                _clear_client_state()
+                _adopt(p, st, "pid recycled")
+            elif action == "maybe_reset":
+                old = cur_before
+                alive = pid_alive(old[0]) if old else False
+                if alive is False:
+                    log("semfix: new client pid %d%s (pid %d gone) - reset "
+                        "per-client pin state"
+                        % (p, tag, old[0] if old else -1))
+                    _clear_client_state()
+                    tracker.resolve(old_dead=True)
+                    _adopt(p, st, "client switch")
+                else:
+                    # NEVER clear the active client's state on the account of
+                    # a newer pid: the m178 hang. Deferred until the old pid
+                    # is confirmed gone (periodic idle probe below).
+                    log("semfix: client pid %d%s appeared while pid %s is "
+                        "alive%s - reset deferred"
+                        % (p, tag, old[0] if old else "?",
+                           "" if alive else " (probe unknown)"))
         for pb in getattr(stream, "new_pbs", []):
             log("stream: new pushbuffer 0x%x" % pb)
             pbs.add(pb)
+        if getattr(stream, "new_pbs", []) and last_pid is not None:
+            progress_t = time.time()
         sems.update(stream.sems)
 
         # The guest-agent trace read is slow under client load; use it only
         # as a fallback for channel/pushbuffer discovery and to recover sems
         # the scanner may have missed.
         if loop % TRACE_EVERY == 0:
+            _hb()
             try:
                 t_sems, t_pbs, t_chans = guest_trace()
             except Exception:
@@ -438,6 +713,7 @@ def main():
             if key != last_state.get(sem):
                 last_state[sem] = key
                 last_seen_t[sem] = now
+                progress_t = time.time()
                 log("uvmt %s" % state_str(sem, st))
             elif now - last_seen_t.get(sem, 0) >= HEARTBEAT:
                 last_seen_t[sem] = now
@@ -473,6 +749,7 @@ def main():
                     seen_sem_pin.add(sem)
                     sem_pin_fail.pop(sem, None)
                     sem_pin_skip_until.pop(sem, None)
+                    progress_t = time.time()
                     if SKIP_PB:
                         continue
                     d = {}
@@ -492,6 +769,7 @@ def main():
                             d["exception"] = str(e)[:40]
                     if res:
                         pin_range_once(*res)
+                        progress_t = time.time()
                         log("semfix: sem=0x%x pushbuf -> %s"
                             % (sem, diag_str(d)))
                     else:
@@ -524,6 +802,7 @@ def main():
             if res:
                 seen_ch.add(ch)
                 pin_range_once(*res)
+                progress_t = time.time()
                 log("semfix: ch=0x%x -> %s" % (ch, diag_str(d)))
             else:
                 log("semfix: ch=0x%x pushbuf miss [%s]" % (ch, diag_str(d)))
@@ -537,12 +816,15 @@ def main():
             if res:
                 seen_pb.add(pb)
                 pin_range_once(*res)
+                progress_t = time.time()
                 log("semfix: pb=0x%x -> %s" % (pb, diag_str(d)))
             else:
                 log("semfix: pb=0x%x pushbuf miss [%s]" % (pb, diag_str(d)))
 
         if DLOG:
-            dmesg_releases(dseen)
+            if dmesg_releases(dseen):
+                progress_t = time.time()
+            _hb()
 
         # Retry incomplete push-buffer pins: gva2gpa of the kernel vmalloc
         # alias fails while the sampled vCPU runs in user mode (PTI), so a
@@ -553,22 +835,83 @@ def main():
                 if not st["done"] and st["tries"] > 0:
                     pin_range_once(*st["args"])
 
+        # Dead-pid detection and the pre-capture stall watchdog. The guest is
+        # probed ONLY when the tracked client has been quiet for a while (a
+        # live, pushing client is never probed), so the agent is not burdened
+        # during healthy runs. A confirmed-dead client is purged immediately:
+        # the NEXT client then starts with clean state - no VM cycle needed
+        # (the kill cascade).
+        now = time.time()
+        if adopted_from_history and last_pid is not None:
+            # The client resumed from pre-restart history may have died while
+            # this process was down; do not latch a dead pid into the counter
+            # retries.
+            alive = pid_alive(last_pid)
+            adopted_from_history = False
+            if alive is False:
+                log("semfix: resumed pid %d already gone - dropping" % last_pid)
+                tracker.note_dead(last_pid, last_start)
+                last_pid = last_start = None
+                _watchdog_reset()
+        if last_pid is not None and now >= probe_next \
+                and now - progress_t >= LIVENESS_IDLE_S:
+            probe_next = now + LIVENESS_EVERY_S
+            alive = pid_alive(last_pid)
+            if alive is False:
+                log("semfix: client pid %d gone (probe) - purging per-client "
+                    "state" % last_pid)
+                tracker.note_dead(last_pid, last_start)
+                _clear_client_state()
+                last_pid = last_start = None
+                _watchdog_reset()
+                if tracker.pending is not None and tracker.current is None \
+                        and tracker.resolve(old_dead=True) == "reset":
+                    p, st = tracker.current
+                    log("semfix: adopting deferred client pid %d" % p)
+                    _adopt(p, st, "deferred reset")
+            elif alive is True and last_pid not in counter_pids:
+                # Pre-capture stall: the client is alive but has no pinned
+                # counter page and nothing progressed. Recover (idempotent),
+                # then escalate loudly if the stall persists - the goal is
+                # that a first cudaMalloc / first GR capture never stalls
+                # silently.
+                idle = now - progress_t
+                if idle >= STALL_PRECAP_S:
+                    if stall_since is None:
+                        stall_since = now
+                    if now >= recover_next:
+                        recover_next = now + STALL_RETRY_S
+                        log("STALL WATCHDOG: client pid %d alive, no "
+                            "pre-capture progress for %.0fs - recovery "
+                            "attempt (re-arm sems, re-pin counter, request "
+                            "pin re-assert)" % (last_pid, idle))
+                        _watchdog_recover()
+                    if not escalated and now - stall_since >= STALL_ESCALATE_S:
+                        escalated = True
+                        log("STALL WATCHDOG: recovery FAILED for pid %d "
+                            "after %.0fs (alive, counter page not pinned) - "
+                            "pre-capture hang persists, needs manual "
+                            "attention" % (last_pid, now - stall_since))
+
         # Pin the completion-counter page of a newly started client, before its
         # first GR pushes decode their releases. The pid comes from the guest
         # scanner; the guest-agent fallback is only a safety net. The exec can
         # block for its whole timeout under client load, so retry on a slow
         # 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()
+            p = cdt_pid()
             _hb()
-            if last_pid:
-                seen_pids.add(last_pid)
-                log("counter: client pid %d (agent)" % last_pid)
+            if p:
+                if tracker.current is None and (p, None) not in tracker.seen:
+                    tracker.observe_history(p, None)
+                last_pid, last_start = p, None
+                log("counter: client pid %d (agent)" % p)
         if last_pid and last_pid not in counter_pids:
             now = time.time()
             if now >= counter_next_try.get(last_pid, 0):
                 try:
-                    pin_counter_page(last_pid, counter_pids)
+                    if pin_counter_page(last_pid, counter_pids):
+                        progress_t = time.time()
                 except Exception as e:
                     log("counter: pin failed: %s" % e)
                 _hb()