#!/usr/bin/env python3
# SPDX-License-Identifier: BSD-3-Clause
# Copyright (c) 2026 Open-IOV contributors
# semfix_uvmt.py - single-QMP tracking-semaphore/resolution watcher.
#
# One QMP client does everything semfix_watch.py and uvmt_poll.py did, because
# QEMU's QMP socket accepts a single client: running the two tools side by side
# silently broke semfix's pinning.
#
# Jobs:
#  1. keep the client's tracking-semaphore pool pages and push-buffer pages
#     pinned with fresh guest truth (semfix_watch.py, imported);
#  2. poll every tracking semaphore seen in the guest trace and print the exact
#     word the client's wait polls: pool GPU VA (gpu_base + idx*4), its guest PA
#     (gva2gpa of the CPU mapping), the payload value and the +0x14/+0x28
#     fields. Every change is appended to a timeline file, so the last wait
#     before a failure and the release that should clear it are both visible;
#  3. mirror the host's decoded releases (dmesg `sema-push ... rel=... pay=...`)
#     into the same timeline, so a wait can be matched to the channel/push that
#     releases its word.
#
# Usage: tools/guest-diag/semfix_uvmt.py [interval-seconds]
# Env:   TIMELINE=/tmp/opencode/uvmt_timeline.log
#        UVMT_HEARTBEAT=30 (seconds between unchanged-state heartbeats)
import os
import re
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
from semfix_watch import (  # noqa: E402
    Qmp, QMP_SOCK, is_kern, is_gpu, diag_str, guest_trace,
    resolve_sem_pool, pin, pin_bulk, resolve_from_channel,
    resolve_from_pushbuffer, resolve_from_sem, resolve_pushbuffer_wide,
    pin_pushbuffer, PUSHBUF_MAX_PAGES,
)

TIMELINE = os.environ.get("TIMELINE", "/tmp/opencode/uvmt_timeline.log")
HEARTBEAT = float(os.environ.get("UVMT_HEARTBEAT", "30"))
DLOG = os.environ.get("UVMT_DMESG", "1") == "1"
VMEXEC = os.path.normpath(os.path.join(HERE, "..", "..", "scripts", "vgpu_vm.py"))
COUNTER_LO = int(os.environ.get("COUNTER_LO", "0x20400f000"), 16)
COUNTER_HI = int(os.environ.get("COUNTER_HI", "0x204010000"), 16)
SEMSTREAM = os.environ.get("SEM_STREAM_FILE", "/tmp/opencode/sem_stream.txt")
TRACE_EVERY = int(os.environ.get("UVMT_TRACE_EVERY", "5"))
# The push-buffer resolution walks the channel/pool/manager chain with many
# QMP reads and can block the poller for tens of seconds. When the GR window
# stream (pin_gr_stream.py) already supplies the push-buffer pages, skip it so
# the tracking-sem poller stays live during the client's deadline window.
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 < 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")
# 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():
    try:
        with open(SEMFIX_HB_FILE, "w") as f:
            f.write("%.3f\n" % time.time())
    except OSError:
        pass


class Stream:
    """Tail the guest scanner's control lines mirrored by pin_gr_stream.py.

    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):
        self.path = path
        self.off = 0
        self.sems = []
        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
        allocates its own channel pools; keeping the old sems/seen-pin marks
        makes the watcher chase stale kernel pointers and pin the previous
        client's (freed) pool pages - the stale-translation trap."""
        self.sems = []
        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:
            with open(self.path) as f:
                f.seek(0, os.SEEK_END)
                size = f.tell()
                if size < self.off:
                    self.off = 0
                f.seek(self.off)
                data = f.read()
                base = self.off
                self.off = f.tell()
        except OSError:
            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:
                    v = int(m.group(1), 16)
                    if v not in self.sems:
                        self.sems.append(v)
                        new_sems.append(v)
            elif line.startswith("#pb"):
                m = re.search(r"0x([0-9a-fA-F]+)", line)
                if m:
                    v = int(m.group(1), 16)
                    if v not in self.pbs:
                        self.pbs.append(v)
                        self.new_pbs.append(v)
            elif line.startswith("#pid"):
                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):
    line = "[%s] %s" % (time.strftime("%H:%M:%S"), msg)
    print(line, flush=True)
    try:
        with open(TIMELINE, "a") as f:
            f.write(line + "\n")
    except OSError:
        pass


def uvmt_state(q, sem):
    """Read one tracking semaphore; returns a dict or None.

    Layout (uvm 570.144, verified by probe): sem+0 = tracking semaphore page
    object, sem+4 low u16 = pool index; page+0 = uvm_rm_mem; mem+0x10 = vas[];
    the payload the client's update_completed_value polls is the u32 at
    cpu_va + idx*4 (its GPU alias is gpu_va + idx*4, the VA the releasing push
    writes).
    """
    vals = q.read64(sem, 12, cache=False)
    if len(vals) < 2 or not is_kern(vals[0]):
        return None
    pv = q.read64(vals[0], cache=False)
    if not pv or not is_kern(pv[0]):
        return None
    mem = pv[0]
    vas = q.read64(mem + 0x10, 16, cache=False)
    cpu = next((v for v in vas if is_kern(v)), None)
    gpu = [v for v in vas if is_gpu(v)]
    idx = vals[1] & 0xffff
    payload = None
    word_pa = None
    if cpu is not None:
        w = q.read64(cpu + idx * 4, cache=False)
        payload = w[0] if w else None
        word_pa = q.gva2gpa(cpu + idx * 4)
    word_va = (gpu[0] + idx * 4) if gpu else None
    return {
        "page": vals[0],
        "mem": mem,
        "idx": idx,
        "gpu": gpu,
        "cpu": cpu,
        "payload": payload,
        "word_va": word_va,
        "word_pa": word_pa,
        "w14": vals[5] if len(vals) > 5 else 0,
        "w28": vals[10] if len(vals) > 10 else 0,
    }


def state_key(st):
    if not st:
        return None
    return (st["idx"], st["payload"], st["w14"], st["w28"], st["word_pa"],
            tuple(st["gpu"]))


def state_str(sem, st):
    if not st:
        return "sem=0x%x unreadable" % sem
    return ("sem=0x%x idx=%d word_va=%s word_pa=%s payload=%s "
            "+14=0x%x +28=0x%x gpu=%s" %
            (sem, st["idx"],
             "0x%x" % st["word_va"] if st["word_va"] is not None else "?",
             "0x%x" % st["word_pa"] if st["word_pa"] is not None else "?",
             "0x%x" % st["payload"] if st["payload"] is not None else "?",
             st["w14"], st["w28"],
             ",".join("0x%x" % v for v in st["gpu"])))


def dmesg_releases(seen):
    """Print/append new host-decoded releases (sema-push lines).

    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",
             "dmesg -T | awk '/sema-svc: thread started/{n=NR} "
             "{l[NR]=$0} END{for(i=n+1;i<=NR;i++) print l[i]}' "
             "| grep 'sema-push' | tail -300"],
            capture_output=True, text=True, timeout=10)
    except Exception:
        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():
    try:
        r = subprocess.run([VMEXEC, "exec", "pgrep -x cdt | head -1"],
                           capture_output=True, text=True, timeout=30)
    except Exception:
        return None
    pids = re.findall(r"(?m)^(\d+)\s*$", r.stdout)
    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.

    The GR channels release the counter page VA range (0x20400fec0..0x20400fff0
    observed); translating it from fresh guest truth before the first releases
    decode is what lets those releases land on the slots the client polls.
    """
    cmd = ("sudo python3 /tmp/guest_pmap_win.py %d 0x%x 0x%x 64"
           % (pid, COUNTER_LO, COUNTER_HI))
    try:
        r = subprocess.run([VMEXEC, "exec", cmd], capture_output=True,
                           text=True, timeout=25)
    except Exception:
        return False
    n = 0
    for m in re.finditer(r"^(0x[0-9a-f]+),(0x[0-9a-f]+),(\d+)$",
                         r.stdout, re.M):
        va, pa, pages = int(m.group(1), 16), int(m.group(2), 16), int(m.group(3))
        if pin_bulk(va, pa, pages * 0x1000):
            n += pages
    if n:
        pinned_pids.add(pid)
        log("counter: pinned %d page(s) at pid %d (0x%x..0x%x)"
            % (n, pid, COUNTER_LO, COUNTER_HI))
    return n > 0


def main():
    interval = float(sys.argv[1]) if len(sys.argv) > 1 else 2.0
    os.environ.setdefault("KEEP_TRACE", "1")
    q = Qmp(QMP_SOCK)
    os.makedirs(os.path.dirname(TIMELINE), exist_ok=True)
    log("semfix_uvmt: watching (interval %.1fs, qmp %s)" % (interval, QMP_SOCK))

    known_sems = set()
    seen_sem_pin = set()
    seen_pb = set()
    seen_ch = set()
    pinned_ranges = {}
    last_state = {}
    last_seen_t = {}
    dseen = set()
    counter_pids = set()
    counter_next_try = {}
    # Current client pid/starttime, mirrored from the tracker (starttime
    # disambiguates recycled pids).
    last_pid = None
    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 = {}
    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)
    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)
        if st is None:
            st = pinned_ranges[key] = {"tries": 0, "done": False,
                                       "args": (gpu_vas, cpu, gpa, size)}
        if st["done"]:
            return
        st["tries"] += 1
        pinned = pin_pushbuffer(q, gpu_vas, cpu, gpa, size)
        expected = min(max(size >> 12, 1) if size else PUSHBUF_MAX_PAGES,
                       PUSHBUF_MAX_PAGES)
        if pinned >= expected:
            st["done"] = True
        elif st["tries"] >= 12:
            st["done"] = True
            log("semfix: pushbuf pin incomplete after %d tries: %d/%d pages"
                % (st["tries"], pinned, expected))

    while True:
        loop += 1
        _hb()
        sems = set()
        pbs = set()
        chans = set()

        new_sems, new_pids, new_nopids = stream.poll()
        for s in new_sems:
            log("stream: new uvmt sem 0x%x" % s)
        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:
                t_sems, t_pbs, t_chans = set(), set(), set()
            _hb()
            sems.update(t_sems)
            pbs.update(t_pbs)
            chans.update(t_chans)

        # Poll every sem ever seen: the client's final wait may only appear in
        # the trace once, and the trace ring wraps while it spins.
        for sem in sorted(sems - known_sems):
            known_sems.add(sem)

        for hb_i, sem in enumerate(sorted(known_sems)):
            if hb_i % 16 == 0:
                # A full pass can run 60-90 s (thousands of QMP reads); keep
                # the heartbeat fresh inside it (r203: 79 s gap -> supervisor
                # restarted the tool mid-pass).
                _hb()
            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)
            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
                log("uvmt(hb) %s" % state_str(sem, st))

            # 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
                    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:
                    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)
                    progress_t = time.time()
                    if SKIP_PB:
                        continue
                    d = {}
                    try:
                        res = resolve_from_sem(q, sem, d)
                    except Exception as e:
                        res = None
                        d["exception"] = str(e)[:40]
                    if not res:
                        for k in ("chunks", "words", "cands", "stop_pool",
                                  "no_gpuva", "no_vmalloc", "empty_vas"):
                            d.pop(k, None)
                        try:
                            res = resolve_pushbuffer_wide(q, sem, d)
                        except Exception as e:
                            res = None
                            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:
                        log("semfix: sem=0x%x pushbuf not found [%s]"
                            % (sem, diag_str(d)))
                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):
            _hb()  # a resolve can block for tens of seconds
            d = {}
            try:
                res = resolve_from_channel(q, ch, d)
            except Exception as e:
                res = None
                d["exception"] = str(e)[:40]
            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)))
        for pb in sorted(pbs - seen_pb):
            _hb()  # a resolve can block for tens of seconds
            d = {}
            try:
                res = resolve_from_pushbuffer(q, pb, d)
            except Exception as e:
                res = None
                d["exception"] = str(e)[:40]
            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:
            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
        # pass can miss a run of pages that resolves a moment later. Only
        # every 8th loop - a full pass costs thousands of QMP reads.
        if loop % 8 == 0:
            for st in list(pinned_ranges.values()):
                if not st["done"] and st["tries"] > 0:
                    _hb()  # pin_range_once can walk thousands of QMP reads
                    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:
            p = cdt_pid()
            _hb()
            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:
                    if pin_counter_page(last_pid, counter_pids):
                        progress_t = time.time()
                except Exception as e:
                    log("counter: pin failed: %s" % e)
                _hb()
                counter_next_try[last_pid] = now + COUNTER_RETRY_S

        time.sleep(interval)


if __name__ == "__main__":
    main()