#!/usr/bin/env python3
"""Cheap per-record watch: the heavy rescan was stalling the guest.

A no-probe control ran 300 s clean while probed runs stalled at 27-255 s
(guest-stalls.md), so the 32 MB rescan every 12 s has to go. After one full
enumeration the craft bases and their roster links are known, and a sample only
needs the hull word at each base -- about 1.2 KB instead of 32 MB. A full rescan
runs only every RESCAN seconds to catch anything genuinely new.
"""
import os, sys, time, struct, collections, importlib.util
SD = __file__.rsplit('/', 1)[0]
sys.path.insert(0, SD)
import gmem, gworld, entities2
_w3 = importlib.util.spec_from_file_location('w3', SD + '/wave3_probe.py')
wave3 = importlib.util.module_from_spec(_w3); _w3.loader.exec_module(wave3)

ROSTER_VT = struct.pack('>I', 0x820AF030)
DELTA, WIN, LINK, HULL = 0x130, 0x400, 0x08, 0x154
BASELINE = 116
# Periodic rescan DISABLED. It was a 32 MB read every 90-180 s, the only
# recurring heavy cost left, and it exists to catch craft that appear from
# nowhere -- which the roster work already showed does not happen: every
# participant is allocated at mission load, so an arrival is a state change on
# an existing craft, not a new allocation. Set >0 to re-enable.
RESCAN = 0

def scan_vt(fd, size, vt):
    out = []
    for a, b in gmem.extents(fd, size):
        pos = a
        while pos < b:
            m = min(1 << 24, b - pos)
            blob = os.pread(fd, m, pos)
            i = blob.find(vt)
            while i != -1:
                if (pos + i) % 4 == 0: out.append(pos + i)
                i = blob.find(vt, i + 1)
            pos += m
    return sorted(out)

def enumerate_craft(fd, defs, want):
    """Full heap scan. Returns [(base, unit, roster_off)].

    Was iterating every 4-byte word of 32 MB in Python -- 8 million steps -- and
    the startup cost worked out at roughly 100 s of wall-clock, which is a third
    of the observation window a turn can afford. Each definition VA is a fixed
    4-byte needle, so bytes.find() does the same search at C speed.
    """
    lo, hi = gmem.va_to_off(entities2.ENT_VA_LO), gmem.va_to_off(entities2.ENT_VA_HI)
    out, pos = [], lo
    while pos < hi:
        m = min(1 << 24, hi - pos)
        blob = os.pread(fd, m, pos)
        for needle, nm in defs.items():
            i = blob.find(needle)
            while i != -1:
                if (pos + i) % 4 == 0:
                    base = pos + i - DELTA
                    head = os.pread(fd, WIN, base)
                    own = None
                    for j in range(0, len(head) - 3, 4):
                        (p,) = struct.unpack_from('>I', head, j)
                        if p in want: own = want[p]; break
                    out.append((base, nm, own))
                i = blob.find(needle, i + 1)
        pos += m
    return sorted(out)

def alive(fd, base):
    try: return struct.unpack('>f', os.pread(fd, 4, base + HULL))[0] > 0
    except Exception: return False

def main():
    secs  = int(sys.argv[1]) if len(sys.argv) > 1 else 300
    every = int(sys.argv[2]) if len(sys.argv) > 2 else 15
    w = gworld.World(); fd = w.fd
    defs = entities2.definitions(w)
    if not defs: print('NOT IN A MISSION'); return 2
    # The 42-instead-of-116 anomaly has now happened twice. Rather than discard
    # immediately, rescan a few times: if the count CLIMBS toward 116 the roster
    # is being built progressively and the probe simply started too early, which
    # is a fact worth having. If it sits at 42 it is something else.
    roster = scan_vt(fd, w.size, ROSTER_VT)
    print('roster records: %d (baseline %d)' % (len(roster), BASELINE))
    for attempt in range(1, 6):
        if len(roster) == BASELINE: break
        time.sleep(10)
        roster = scan_vt(fd, w.size, ROSTER_VT)
        print('  rescan %d: %d records' % (attempt, len(roster)), flush=True)
    if len(roster) != BASELINE:
        print('DISCARD: settled at %d, not the reproduced baseline %d'
              % (len(roster), BASELINE)); return 3
    f = os.fdopen(os.dup(fd), 'rb')
    label, want = {}, {}
    for o in roster:
        va = gmem.primary_va(o)
        if va is None: continue
        label[o] = wave3.resolve_id(f, w.size, o)[0] or '?'
        want[va + LINK] = o

    # A 1 MB window was too narrow and found nothing, yet the summary still
    # printed "stalled samples=0" -- a witness that does not exist cannot report
    # zero stalls, and that is false reassurance, not a clean run. Search wider,
    # and if there is still no witness say the run is UNVALIDATED.
    # Taking the FIRST word that advances was wrong: it flagged 13 samples as
    # stalled in a run that was recording losses in those same samples, which a
    # frozen guest cannot do. Some counters simply advance intermittently. Use
    # timer_probe's approach instead -- collect every candidate, keep the modal
    # rate cluster, and call a stall only when a MAJORITY of that cluster fails
    # to advance.
    # 4 MB was too narrow: it yielded 5 candidates at a nominal 24/s that then
    # failed to advance in 15 s windows where craft were being destroyed, i.e.
    # they are bursty, not frame counters. timer_probe searched the WHOLE 32 MB
    # region and found 286 with a clean cluster at ~17/s. Pay the one-off cost.
    # REVERTED to the full-region calibration. Trimming it to 8 spread 512 KB
    # windows to save I/O found only 17 candidates, none of them frame counters,
    # and the witness then reported 0/17 stalled on EVERY sample of a run that
    # recorded 13 losses -- a total contradiction. The frame-rate cluster is
    # sparse and spread sampling misses it. Two 32 MB reads once at startup is
    # the price of a witness that works; the recurring cost was the periodic
    # rescan, and that is what has been cut instead.
    lo, hiw = gmem.va_to_off(0xBC000000), gmem.va_to_off(0xBE000000)
    def grab():
        out, pos = bytearray(), lo
        while pos < hiw:
            n = min(1 << 24, hiw - pos); out += os.pread(fd, n, pos); pos += n
        return bytes(out)
    a = grab(); time.sleep(3.0); b = grab()
    # The candidate search was the last Python loop over 8 million words and cost
    # most of the remaining startup time. numpy does the same comparison as two
    # vector ops. (AGENT.md: numpy is installed; its absence used to look like a
    # logic bug.)
    import numpy as np
    n4 = min(len(a), len(b)) // 4 * 4
    A = np.frombuffer(a[:n4], dtype='>u4').astype(np.int64)
    B = np.frombuffer(b[:n4], dtype='>u4').astype(np.int64)
    d = B - A
    rate = d / 3.0
    idx = np.nonzero((d > 0) & (rate > 5) & (rate < 200))[0]
    cands = [(lo + int(i) * 4, int(round(float(rate[i])))) for i in idx]
    # The MODAL cluster is not the frame counter. One run picked a modal rate of
    # 93/s, and only 11 of 31 of those advanced during active combat -- they are
    # subsystem counters that tick in bursts. timer_probe measured the frame-rate
    # cluster at ~17/s, matching Canary's 14-19 fps on this box, so prefer a
    # cluster in that band and fall back to modal only if none exists.
    rates = collections.Counter(r for _, r in cands)
    band = [r for r in rates if 8 <= r <= 40]
    pick = max(band, key=lambda r: rates[r]) if band else (
        rates.most_common(1)[0][0] if rates else None)
    ticks = [o for o, r in cands if r == pick][:32]
    if len(ticks) < 8:
        print('WARNING: only %d witnesses; stall detection is weak' % len(ticks))
    print('tick witnesses: %d candidates, using %d at %s/s (frame-rate band)'
          % (len(cands), len(ticks), pick)
          if ticks else 'tick witnesses: NONE -- RUN UNVALIDATED')
    last_ticks = [struct.unpack('>I', os.pread(fd, 4, o))[0] for o in ticks]

    craft = enumerate_craft(fd, defs, want)
    print('craft %d, linked %d' % (len(craft), sum(1 for c in craft if c[2])))
    prev = collections.Counter(c[2] for c in craft if c[2] and alive(fd, c[0]))
    print('t=   0s deployed=%d strengths %s'
          % (len(prev), sorted(collections.Counter(prev.values()).items())), flush=True)
    t0 = time.time(); last_rescan = t0; arr = los = 0; stalls = 0
    pending, confirmed = {}, 0
    while time.time() - t0 < secs:
        time.sleep(every)
        el = round(time.time() - t0)
        if RESCAN and time.time() - last_rescan > RESCAN:
            craft = enumerate_craft(fd, defs, want); last_rescan = time.time()
        cur = collections.Counter(c[2] for c in craft if c[2] and alive(fd, c[0]))
        st = ''
        if ticks:
            now = [struct.unpack('>I', os.pread(fd, 4, o))[0] for o in ticks]
            moved = sum(1 for x, y in zip(last_ticks, now) if y > x)
            # "Fewer than half" was too strict: 11 of 31 advanced while craft
            # were being destroyed. The unambiguous signal in that run was
            # 0 of 31, which coincided exactly with losses stopping.
            if moved == 0:
                st = '  *** GUEST STALLED (%d/%d witnesses moved) ***' % (moved, len(ticks))
                stalls += 1
            last_ticks = now
        # Report EVERY increase, not just 0 -> n. The first run logged a record
        # reading 13 then 14 with no event printed, which is how flicker in the
        # hull-based liveness read hides: only decreases were being surfaced, so
        # a spurious 0 -> 2 looked like an arrival while 13 -> 14 looked like
        # nothing. An arrival must also PERSIST to count.
        a_ = [(o, prev.get(o, 0), cur[o]) for o in cur if cur[o] > prev.get(o, 0)]
        l_ = [(o, prev[o], cur.get(o, 0)) for o in prev if cur.get(o, 0) < prev[o]]
        for o, x, y in a_:
            if x == 0: pending[o] = pending.get(o, 0) + 1
        for o in list(pending):
            if cur.get(o, 0) == 0: pending.pop(o, None)          # vanished: flicker
            elif pending[o] == 2:
                confirmed += 1
                print('   *** CONFIRMED ARRIVAL %-26s now %d (persisted 2 samples)'
                      % (label.get(o, '?'), cur[o]), flush=True)
                pending[o] = 3
        arr += sum(1 for x in a_ if x[1] == 0); los += len(l_)
        print('t=%4ds deployed=%d up=%d down=%d (cum up %d / down %d, confirmed %d)%s'
              % (el, len(cur), len(a_), len(l_), arr, los, confirmed, st), flush=True)
        for o, x, y in a_:
            print('   up      %-30s %d -> %d%s' % (label.get(o, '?'), x, y,
                  '   <- candidate arrival' if x == 0 else ''), flush=True)
        for o, x, y in l_: print('   loss    %-30s %d -> %d' % (label.get(o, '?'), x, y), flush=True)
        prev = cur
    if not ticks:
        print('\nRUN UNVALIDATED: no tick witness, so flat samples prove nothing.')
    print('\nTOTAL candidate-up=%d down=%d CONFIRMED arrivals=%d stalled samples=%s'
          % (arr, los, confirmed, stalls if ticks else 'UNKNOWN'))
    return 0

if __name__ == '__main__':
    sys.exit(main())