#!/usr/bin/env python3 """Shared harness for live-memory probes. Written because the same four lessons were re-learned in four separate one-off probes, each time by losing a run: * **stall witness** — flat output is indistinguishable from a frozen guest. Four probes shipped without one; three produced runs that could not be interpreted (guest-stalls.md, remaining-ob-hunt.md). * **incremental save** — two probes deferred results to the end and a turn timeout destroyed them, the second time four iterations after the lesson was written down. * **baseline discard** — a run whose roster count is not the stage's member count is a different stage and must not be interpreted (mission-per-record-strength.md). * **witness first** — a run's stall state must be read before its numbers. Anything reading live guest memory should start from `Probe`, not from a blank file. p = Probe(baseline=116) # None = accept whatever loads if not p.ok: sys.exit(p.why) p.log('t\\tvalue') # opens an incrementally-flushed TSV while p.tick(every=15): p.emit('%d\\t%d' % (p.elapsed, measure())) print(p.status()) # includes the stall verdict """ import collections, os, struct, sys, time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import gmem, gworld, entities2 import numpy as np ROSTER_VT = struct.pack('>I', 0x820AF030) DELTA, WIN, LINK, HULL = 0x130, 0x400, 0x08, 0x154 HEAP_LO, HEAP_HI = 0xBC000000, 0xBE000000 ENT_LO, ENT_HI = entities2.ENT_VA_LO, entities2.ENT_VA_HI class Probe: def __init__(self, baseline=None, tsv=None, witness=True, rescans=5): self.w = gworld.World() self.fd = self.w.fd self.t0 = time.time() self.ok, self.why = False, '' self.stalled_samples = 0 self._f = None self.defs = entities2.definitions(self.w) if not self.defs: self.why = 'NOT IN A MISSION (no unit definitions)' return self.roster = self._scan_vt(ROSTER_VT) for _ in range(rescans): if baseline is None or len(self.roster) == baseline: break time.sleep(10) self.roster = self._scan_vt(ROSTER_VT) if baseline is not None and len(self.roster) != baseline: self.why = ('DISCARD: roster settled at %d, not the baseline %d ' '(a different stage loaded)' % (len(self.roster), baseline)) return self.want = {} for o in self.roster: va = gmem.primary_va(o) if va is not None: self.want[va + LINK] = o self.ticks, self.rate, self._last = [], None, None if witness: self._calibrate() if tsv: self.log_path = tsv self.ok = True # ---- guest access ------------------------------------------------- def _scan_vt(self, vt): out = [] for a, b in gmem.extents(self.fd, self.w.size): pos = a while pos < b: m = min(1 << 24, b - pos) blob = os.pread(self.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 heap(self): """Raw bytes of the game heap.""" lo, hi = gmem.va_to_off(HEAP_LO), gmem.va_to_off(HEAP_HI) out, pos = bytearray(), lo while pos < hi: n = min(1 << 24, hi - pos) out += os.pread(self.fd, n, pos) pos += n return bytes(out), lo def craft(self): """[(base_offset, unit_name, roster_offset|None)] — one per live craft.""" lo, hi = gmem.va_to_off(ENT_LO), gmem.va_to_off(ENT_HI) out, pos = [], lo while pos < hi: m = min(1 << 24, hi - pos) blob = os.pread(self.fd, m, pos) for needle, nm in self.defs.items(): i = blob.find(needle) while i != -1: if (pos + i) % 4 == 0: base = pos + i - DELTA own = None head = os.pread(self.fd, WIN, base) for j in range(0, len(head) - 3, 4): (p,) = struct.unpack_from('>I', head, j) if p in self.want: own = self.want[p] break out.append((base, nm, own)) i = blob.find(needle, i + 1) pos += m return out def alive(self, base): try: return struct.unpack('>f', os.pread(self.fd, 4, base + HULL))[0] > 0 except Exception: return False def strengths(self): """craft alive per roster record.""" per = collections.Counter() for base, _, own in self.craft(): if own is not None and self.alive(base): per[own] += 1 return per # ---- witness ------------------------------------------------------ def _calibrate(self, dt=3.0): a, _ = self.heap() time.sleep(dt) b, _ = self.heap() A = np.frombuffer(a, dtype='>u4').astype(np.int64) B = np.frombuffer(b, dtype='>u4').astype(np.int64) d = B - A rates = collections.Counter((d[(d > 15) & (d < 600)] // int(dt)).tolist()) band = [r for r in rates if 8 <= r <= 40] pick = max(band, key=lambda r: rates[r]) if band else None if pick is None: return self.rate = pick self.ticks = np.nonzero(d // int(dt) == pick)[0][:32] self._last = B[self.ticks] def stalled(self): """True if the guest did not advance since the previous check.""" if not len(self.ticks): return None # unknown: no witness b, _ = self.heap() now = np.frombuffer(b, dtype='>u4').astype(np.int64)[self.ticks] moved = int((now > self._last).sum()) self._last = now return moved == 0 # ---- loop / output ------------------------------------------------ @property def elapsed(self): return round(time.time() - self.t0) def tick(self, every, secs): if time.time() - self.t0 >= secs: return False time.sleep(every) self._stall = self.stalled() if self._stall: self.stalled_samples += 1 return True def status(self): s = getattr(self, '_stall', None) if s is None: return ' (UNVALIDATED: no witness)' return ' *** GUEST STALLED ***' if s else '' def log(self, header, path=None): self._f = open(path or getattr(self, 'log_path', '/tmp/probe.tsv'), 'w') self._f.write(header.rstrip('\n') + '\n') self._f.flush() def emit(self, line): """Write AND flush — never defer results to the end of a run.""" if self._f: self._f.write(line.rstrip('\n') + '\n') self._f.flush() def summary(self): w = ('%d witnesses at %s/s' % (len(self.ticks), self.rate) if len(self.ticks) else 'NO WITNESS - run UNVALIDATED') return ('roster %d, definitions %d, %s, stalled samples %d' % (len(self.roster), len(self.defs), w, self.stalled_samples))