re: build a shared probe harness so the same lessons stop being re-learned

Four probes were written from a blank file and each re-learned the same lessons
by losing a run: that a flat run cannot be told from a frozen guest without a
stall witness, that results held to the end of a run are destroyed by a turn
timeout, that a roster count which is not the stage's member count means a
different stage loaded and must be discarded, and that a run's witness state has
to be read before its numbers. Writing each lesson down did not stop the next
probe repeating it, because each probe started from nothing.

probeharness.py makes them structural. Probe(baseline=N) discovers the roster,
rescans up to five times and refuses to start if the count never reaches the
baseline. The witness is calibrated on construction, sampled by tick() and
reported by status() and summary(), so a probe cannot forget it, and when no
witness is found it reports UNVALIDATED rather than zero stalls. emit() flushes
on every line. craft(), strengths(), alive() and heap() supply the
roster-to-craft link, per-record liveness and the raw heap, so a new probe
writes only its own logic.

Verified rather than asserted: deploy_probe.py reimplements the per-record
deployment watch on top of it in about forty lines against wave7_probe's
hundred and fifty, and its first live run was clean -- 116 roster records, 32
witnesses at 10/s, zero stalled samples, seven losses tracked, and the TSV
written incrementally. Nothing about the result is new, which is the point: the
harness reproduces a known-good measurement.

The existing probes are deliberately not ported. They work, and rewriting them
would risk changing results other documents cite. New probes should use the
harness; old ones should be ported when they next need a change.

docs/re/BACKLOG.md

@@ -630,6 +630,18 @@ search cannot find a *schedule*.
   that cannot be told from a freeze**; the recurring fix is the shared harness
   noted earlier. Needs two HUD readings at *different* values in non-stalled
   samples.
+* ✅ **(2026-08-25) SHARED PROBE HARNESS built and verified**
+  ([`probe-harness.md`](probe-harness.md), `tools/re-capture/probeharness.py`).
+  Makes structural the four lessons that were re-learned in four separate probes:
+  **built-in stall witness** (says `UNVALIDATED` when absent rather than reporting
+  zero stalls), **`emit()` flushes every line** so a timeout cannot destroy
+  results, **baseline discard with rescan** (`Probe(baseline=116)` refuses to
+  start on a different stage), and `summary()` putting the witness first.
+  Also provides `craft()`/`strengths()`/`alive()`/`heap()` so a new probe writes
+  only its own logic. ✅ Verified: `deploy_probe.py` reimplements the deployment
+  watch in ~40 lines vs 150, first live run clean — 116 roster, 32 witnesses at
+  10/s, **0 stalled**, 7 losses, TSV written incrementally. ⚠️ Existing probes
+  deliberately **not** ported — they work and other docs cite their results.
 * ~~🚧 BLOCKER: t=210/240 unreachable in one turn~~ — **superseded, see above**;
   it rested on an untested assumption that a turn is one shell call. 595 s shell cap − ~220 s boot (a ~190 s title movie that cannot be
   tapped through) − ~25 s startup = **~350 s observation ≈ 193 game-seconds**.

docs/re/probe-harness.md

@@ -0,0 +1,61 @@
+# A shared probe harness — so the same lessons stop being re-learned
+
+Status: ✅ built and verified on a live run.
+
+## Why
+
+Four separate probes were written from a blank file, and each re-learned the same
+lessons by losing a run:
+
+| lesson | where it was learned | where it was re-learned |
+|---|---|---|
+| a flat run is indistinguishable from a frozen guest without a **stall witness** | `guest-stalls.md` | 3 further probes |
+| **save incrementally** — a turn timeout destroys anything held to the end | `guest-stalls.md` | `ob_probe2.py`, four iterations later |
+| a roster count that is not the stage's member count is **a different stage** — discard | `mission-per-record-strength.md` | — |
+| **read the witness first**, before interpreting any number | `remaining-ob-hunt.md` | — |
+
+Writing each lesson down did not stop the next probe from repeating it, because
+each probe started from nothing. `tools/re-capture/probeharness.py` makes them
+structural instead.
+
+## What it provides
+
+```python
+p = Probe(baseline=116)              # None = accept whatever loads
+if not p.ok: sys.exit(p.why)         # e.g. "DISCARD: roster settled at 42"
+print(p.summary())                   # witness count + rate, read this FIRST
+p.log('t\tvalue')                    # opens a TSV, flushed on every emit
+while p.tick(every=15, secs=300):
+    p.emit(...)                      # written and flushed immediately
+    print(p.status())                # '' | '*** GUEST STALLED ***' | UNVALIDATED
+```
+
+* `Probe(baseline=…)` discovers the roster, **rescans up to five times**, and
+  refuses to start if the count never reaches the baseline.
+* The **witness is built in**: calibrated on construction, sampled by `tick()`,
+  reported by `status()` and `summary()`. A probe cannot forget it, and when no
+  witness is found it says `UNVALIDATED` rather than reporting zero stalls.
+* `emit()` **flushes on every line** — results cannot be lost to a timeout.
+* `craft()`, `strengths()`, `alive()` and `heap()` provide the roster→craft link,
+  per-record liveness and the raw heap, so a new probe writes only its own logic.
+
+## ✅ Verified
+
+`deploy_probe.py` reimplements the per-record deployment watch on top of it, in
+about 40 lines against `wave7_probe.py`'s 150. First live run:
+
+```
+roster 116, definitions 14, 32 witnesses at 10/s, stalled samples 0
+t= 249s deployed=40 up=0 down=1 (cum 0/7)
+```
+
+Baseline check passed, witness found and reporting, seven losses tracked,
+`/tmp/deploy.tsv` written incrementally. Nothing about the result is new — that
+is the point; the harness reproduces a known-good measurement.
+
+## Not done
+
+The existing probes (`wave6/7`, `ob_probe2`, `ob_by_hud`, `census`, `liveness`)
+are **not** ported. They work, and rewriting them risks changing results that
+other documents cite. New probes should use the harness; old ones should be
+ported only when they next need a change.

tools/re-capture/deploy_probe.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+"""Per-record deployment watch, built on the shared harness.
+
+Replaces wave7_probe.py's hand-rolled setup. Everything that used to be
+re-invented per probe -- roster discovery, the baseline discard rule, the craft
+link, the stall witness, incremental output -- comes from `probeharness.Probe`.
+"""
+import collections, sys
+sys.path.insert(0, __file__.rsplit('/', 1)[0])
+from probeharness import Probe
+
+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
+    base  = int(sys.argv[3]) if len(sys.argv) > 3 else 116
+
+    p = Probe(baseline=base, tsv='/tmp/deploy.tsv')
+    if not p.ok:
+        print(p.why)
+        return 3
+    print(p.summary(), flush=True)
+    p.log('t\tdeployed\tcraft\tstalled')
+
+    prev = p.strengths()
+    print('t=   0s deployed=%d strengths %s'
+          % (len(prev), sorted(collections.Counter(prev.values()).items())), flush=True)
+    ups = downs = 0
+    while p.tick(every, secs):
+        cur = p.strengths()
+        up   = [(o, prev.get(o, 0), cur[o]) for o in cur if cur[o] > prev.get(o, 0)]
+        down = [(o, prev[o], cur.get(o, 0)) for o in prev if cur.get(o, 0) < prev[o]]
+        ups += len(up); downs += len(down)
+        print('t=%4ds deployed=%d up=%d down=%d (cum %d/%d)%s'
+              % (p.elapsed, len(cur), len(up), len(down), ups, downs, p.status()),
+              flush=True)
+        p.emit('%d\t%d\t%d\t%s'
+               % (p.elapsed, len(cur), sum(cur.values()), p.status().strip()))
+        prev = cur
+    print('\n%s' % p.summary())
+    print('TOTAL up=%d down=%d' % (ups, downs))
+    return 0
+
+if __name__ == '__main__':
+    sys.exit(main())

tools/re-capture/deploy_session.sh

@@ -0,0 +1,31 @@
+#!/usr/bin/env bash
+set -u
+export HOME=/sylph-home/re SDL_AUDIODRIVER=dummy DISPLAY=:98
+export PYTHONPATH=/sylph-home/.local/lib/python3.12/site-packages
+SD="$(cd "$(dirname "$0")" && pwd)"
+SECS="${1:-180}"; EVERY="${2:-10}"; HUNT="${3:-1}"
+CFG=/tmp/nav-live.json
+"$SD/launch_mission.sh" fly || { echo "BOOT FAILED"; exit 1; }
+# The bind is intermittent and a failed bind means no pilot, no kills and a
+# completely uninformative run. Retry before giving up, and abort if it never
+# takes rather than silently flying an unattended craft.
+# entities2 self finds the player by MOTION between two samples, so a craft that
+# is sitting still at mission start is invisible and the bind fails -- three
+# times in a row on the run that added this retry. Nudge the throttle first so
+# there is something to see, then bind.
+BOUND=0
+for try in 1 2 3; do
+  python3 "$SD/pad.py" set "rt=1" >/dev/null 2>&1 || true
+  sleep 3
+  python3 "$SD/pad.py" clear >/dev/null 2>&1 || true
+  if python3 "$SD/entities2.py" self 0x130 "$CFG" >/dev/null 2>&1; then BOUND=1; break; fi
+  echo "--- bind attempt $try failed, retrying"; sleep 5
+done
+if [ "$BOUND" = 1 ]; then
+  SYLPH_HUNT="$HUNT" SYLPH_KILL_TURRETS=1 SYLPH_KEEPOUT="${SYLPH_KEEPOUT:-1400}" SYLPH_HZ="${SYLPH_HZ:-8.0}" SYLPH_PREFER="${SYLPH_PREFER:-}" nohup python3 "$SD/pilot.py" "$CFG" "$SECS" \
+      </dev/null >/tmp/live-pilot.log 2>&1 &
+  PILOT=$!; echo "--- pilot (SYLPH_HUNT=$HUNT)"
+else echo "BIND FAILED after 3 attempts -- aborting, an unattended run tests nothing"; exit 4; fi
+python3 "$SD/deploy_probe.py" "$SECS" "$EVERY"; rc=$?
+[ -n "$PILOT" ] && kill "$PILOT" 2>/dev/null
+echo "LIVENESS DONE rc=$rc"

tools/re-capture/probeharness.py

@@ -0,0 +1,199 @@
+#!/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))