code wiki / _hdl_build / nx_vqoe_gate.nx
`nx_vqoe_gate.nx` source

↩ module page · 373 lines · 15797 B
1// nx_vqoe_gate.nx -- the STANDING VIDEO SHIP-GATE (F601; eats debt seq35).
2// WHY: prior /video seeds measured the ENCODER VM (36-81fps) while the received room ran 1fps --
3// the 100x engine-truth-vs-field-truth gap that made /video suck. This organ grades the thing the
4// family actually experiences: the RECEIVER-side /api/vqoe beacons (vqoe.jsonl, appended by
5// nx_mgmt_api ma_vqoe from real rooms + the headless probe). It publishes a machine verdict the
6// reliability plane (nx_deploy_ready via knowledge/registry/deploy_checks.tsv) and any agent can
7// gate on BEFORE shipping video changes.
8// VERDICTS (honest three-state):
9//   GREEN = fresh beacons within thresholds -> ship.
10//   AMBER = insufficient fresh evidence -> ship is UNPROVEN (not blocked; the remediation is to
11//           generate evidence: headless probe or a real call). No GREEN marker = deploy_ready warns.
12//   RED   = received-room truth failed a threshold -> DO NOT ship video changes.
13// DATA-DRIVEN (rule 11/17): thresholds live in knowledge/registry/vqoe_gate.conf key=value rows,
14// compiled defaults only as bootstrap fallback. ENVELOPE DECLARED (scale law): reads the TAIL
15// tail_bytes of vqoe.jsonl (default 4MiB), keeps <=8192 rtt samples; over-cap is COUNTED + printed,
16// never silent. Imports ONLY nx_syscalls = drift-immune (the reliability-guard pattern).
17//   usage: nx_vqoe_gate [jsonl] [conf] [outlog]   (defaults: vqoe.jsonl,
18//          knowledge/registry/vqoe_gate.conf, knowledge/status/video_vqoe_gate.log)
19// license_tier: ORIGINAL
20import "nx_syscalls.nx"
21
22const VG_RTT_CAP: i64 = 8192
23
24func vg_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
25
26func vg_s(dst: *u8, off: i64, s: *u8) -> i64 {
  var i: i64 = 0
  while s[i] != (0 as u8) { dst[off+i] = s[i]; i = i + 1 }
  return off + i
30}
31
32func vg_n(dst: *u8, off: i64, v: i64) -> i64 {
  var m: i64 = v
  var o: i64 = off
  if m < 0 { dst[o] = 45 as u8; o = o + 1; m = 0 - m }
  let t: *u8 = sys_mmap(32)
  var k: i64 = 0
  if m == 0 { t[0] = 48 as u8; k = 1 }
  while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
  var j: i64 = 0
  while j < k { dst[o+j] = t[k-1-j]; j = j + 1 }
  return o + k
43}
44
45func vg_starts(b: *u8, off: i64, le: i64, lit: *u8) -> i64 {
  var i: i64 = 0
  while lit[i] != (0 as u8) {
      if off + i >= le { return 0 }
      if b[off+i] != lit[i] { return 0 }
      i = i + 1
  }
  return 1
53}
54
55// find `needle` in [start,le); parse the (signed) integer right after it. Returns position
56// after the value (for repeat scans) or -1 if not found. Value -> oval[0].
57func vg_findint(b: *u8, le: i64, needle: *u8, start: i64, oval: *i64) -> i64 {
  var i: i64 = start
  while i < le {
      if vg_starts(b, i, le, needle) == 1 {
          var nl: i64 = 0
          while needle[nl] != (0 as u8) { nl = nl + 1 }
          var p: i64 = i + nl
          var neg: i64 = 0
          if p < le { if b[p] == (45 as u8) { neg = 1; p = p + 1 } }
          var v: i64 = 0
          var any: i64 = 0
          var sc: i64 = 1
          while sc == 1 {
              if p >= le { sc = 0 } else {
                  let c: i64 = b[p] as i64
                  if c < 48 { sc = 0 } else { if c > 57 { sc = 0 } else { v = v * 10 + (c - 48); any = 1; p = p + 1 } }
              }
          }
          if any == 0 { oval[0] = 0 - 1; return p }
          if neg == 1 { v = 0 - v }
          oval[0] = v
          return p
      }
      i = i + 1
  }
  return 0 - 1
83}
84
85// conf line "<key>=<uint>" -> value, else -1
86func vg_lineval(cb: *u8, ls: i64, le: i64, key: *u8) -> i64 {
  var k: i64 = 0
  while key[k] != (0 as u8) {
      if ls + k >= le { return 0 - 1 }
      if cb[ls+k] != key[k] { return 0 - 1 }
      k = k + 1
  }
  if ls + k >= le { return 0 - 1 }
  if cb[ls+k] != (61 as u8) { return 0 - 1 }
  var v: i64 = 0
  var any: i64 = 0
  var p: i64 = ls + k + 1
  while p < le {
      let c: i64 = cb[p] as i64
      if c >= 48 { if c <= 57 { v = v * 10 + (c - 48); any = 1 } }
      p = p + 1
  }
  if any == 0 { return 0 - 1 }
  return v
105}
106
107func vg_confget(cb: *u8, cn: i64, key: *u8, dflt: i64) -> i64 {
  if (cb as i64) == 0 { return dflt }
  var i: i64 = 0
  while i < cn {
      var e: i64 = i
      var sc: i64 = 1
      while sc == 1 {
          if e >= cn { sc = 0 } else { if cb[e] == (10 as u8) { sc = 0 } else { e = e + 1 } }
      }
      if e > i { if cb[i] != (35 as u8) {
          let v: i64 = vg_lineval(cb, i, e, key)
          if v >= 0 { return v }
      } }
      i = e + 1
  }
  return dflt
123}
124
125func main(argc: i64, argv: *i64) -> i64 {
  var jsonl: *u8 = "vqoe.jsonl" as *u8
  var confp: *u8 = "knowledge/registry/vqoe_gate.conf" as *u8
  var outp: *u8 = "knowledge/status/video_vqoe_gate.log" as *u8
  if argc >= 2 { jsonl = argv[1] as *u8 }
  if argc >= 3 { confp = argv[2] as *u8 }
  if argc >= 4 { outp = argv[3] as *u8 }
132
  // ---- thresholds (data-driven; compiled numbers are BOOTSTRAP DEFAULTS only) ----
  let clen: *i64 = sys_mmap(8) as *i64
  let cb: *u8 = sys_read_file(confp, clen)
  var cn: i64 = 0
  if (cb as i64) != 0 { cn = clen[0] }
  let window: i64 = vg_confget(cb, cn, "window_sec" as *u8, 86400)
  let minb: i64 = vg_confget(cb, cn, "min_beacons" as *u8, 3)
  let decmin: i64 = vg_confget(cb, cn, "dec_fps_min" as *u8, 8)
  let rttmax: i64 = vg_confget(cb, cn, "rtt_ms_max" as *u8, 2000)
  let dropmax: i64 = vg_confget(cb, cn, "drop_pct_max" as *u8, 20)
  let tailb: i64 = vg_confget(cb, cn, "tail_bytes" as *u8, 4194304)
  let starvmax: i64 = vg_confget(cb, cn, "starv_pct_max" as *u8, 25)
145
  // ---- now (wall clock) ----
  let tsb: *i64 = sys_mmap(16) as *i64
  tsb[0] = 0
  sys_clock_gettime_real(tsb)
  let now: i64 = tsb[0]
151
  // ---- tail-read the beacon journal (ENVELOPE: last tail_bytes only, DECLARED below) ----
  var jb: *u8 = 0 as *u8
  var jn: i64 = 0
  var ds: i64 = 0
  var fsz: i64 = 0
  let fd: i64 = sys_openat_rd(jsonl)
  if fd >= 0 {
      fsz = sys_lseek(fd, 0, 2)
      var start: i64 = 0
      if fsz > tailb { start = fsz - tailb }
      sys_lseek(fd, start, 0)
      jb = sys_mmap(tailb + 64)
      var got: i64 = 0
      var rsc: i64 = 1
      while rsc == 1 {
          let r: i64 = sys_read(fd, jb + got, tailb - got)
          if r <= 0 { rsc = 0 } else { got = got + r; if got >= tailb { rsc = 0 } }
      }
      jn = got
      sys_close(fd)
      if start > 0 {
          var ssc: i64 = 1
          while ssc == 1 {
              if ds >= jn { ssc = 0 } else { if jb[ds] == (10 as u8) { ssc = 0; ds = ds + 1 } else { ds = ds + 1 } }
          }
      }
  }
179
  // ---- scan beacons ----
  let rtts: *i64 = sys_mmap(VG_RTT_CAP * 8) as *i64
  var nrtt: i64 = 0
  let decs: *i64 = sys_mmap(VG_RTT_CAP * 8) as *i64
  var ndec: i64 = 0
  var overcap: i64 = 0
  var nb: i64 = 0
  var nprobe: i64 = 0
  var nstarv: i64 = 0
  var newest: i64 = 0
  var worstdec: i64 = 99999
  var dksum: i64 = 0
  var rxsum: i64 = 0
  let vbox: *i64 = sys_mmap(8) as *i64
  var i: i64 = ds
  while i < jn {
      var e: i64 = i
      var sc: i64 = 1
      while sc == 1 {
          if e >= jn { sc = 0 } else { if jb[e] == (10 as u8) { sc = 0 } else { e = e + 1 } }
      }
      if e > i + 8 {
          let tp: i64 = vg_findint(jb, e, "\"t\":" as *u8, i, vbox)
          if tp >= 0 {
              let t: i64 = vbox[0]
              if t > newest { newest = t }
              if t >= now - window { if t <= now + 3600 {
                  // 2026-07-29: DIAGNOSTIC HARNESS ROWS ARE NOT FAMILY TRUTH. Build 855+ tags probe
                  // beacons "tp":1 (?probe=1 auto-join rooms). Grade REAL rows only; count the
                  // excluded loudly. (Pre-855 probe rows are unmarked -- they age out of the window.)
                  var isprobe: i64 = 0
                  let pbp: i64 = vg_findint(jb, e, "\"tp\":" as *u8, i, vbox)
                  if pbp >= 0 { if vbox[0] == 1 { isprobe = 1 } }
                  if isprobe == 1 { nprobe = nprobe + 1 } else {
                  nb = nb + 1
                  let rp: i64 = vg_findint(jb, e, "\"rtt\":" as *u8, i, vbox)
                  if rp >= 0 { if vbox[0] >= 0 {
                      if nrtt < VG_RTT_CAP { rtts[nrtt] = vbox[0]; nrtt = nrtt + 1 } else { overcap = overcap + 1 }
                  } }
                  var sawneg: i64 = 0
                  var dp: i64 = i
                  var dsc: i64 = 1
                  while dsc == 1 {
                      dp = vg_findint(jb, e, "\"dec\":" as *u8, dp, vbox)
                      if dp < 0 { dsc = 0 } else {
                          if vbox[0] >= 0 {
                              if vbox[0] < worstdec { worstdec = vbox[0] }
                              if ndec < VG_RTT_CAP { decs[ndec] = vbox[0]; ndec = ndec + 1 } else { overcap = overcap + 1 }
                          } else { sawneg = 1 }
                      }
                  }
                  var kp: i64 = i
                  var ksc: i64 = 1
                  while ksc == 1 {
                      kp = vg_findint(jb, e, "\"dk\":" as *u8, kp, vbox)
                      if kp < 0 { ksc = 0 } else { if vbox[0] > 0 { dksum = dksum + vbox[0] } }
                  }
                  var sawrx: i64 = 0
                  var xp: i64 = i
                  var xsc: i64 = 1
                  while xsc == 1 {
                      xp = vg_findint(jb, e, "\"rx\":" as *u8, xp, vbox)
                      if xp < 0 { xsc = 0 } else { if vbox[0] > 0 { rxsum = rxsum + vbox[0]; sawrx = 1 } }
                  }
                  // DECODE-STARVATION (the probe835 blind spot 2026-07-20): frames ARRIVING (rx>0) but the
                  // decoder NEVER producing (dec=-1) is a hard failure signature, not a skip -- count it.
                  if sawneg == 1 { if sawrx == 1 { nstarv = nstarv + 1 } }
                  }
              } }
          }
      }
      i = e + 1
  }
253
  // ---- p50 rtt (insertion sort; nrtt <= 8192 by construction) ----
  var si: i64 = 1
  while si < nrtt {
      let key: i64 = rtts[si]
      var sj: i64 = si - 1
      var msc: i64 = 1
      while msc == 1 {
          if sj < 0 { msc = 0 } else { if rtts[sj] <= key { msc = 0 } else { rtts[sj+1] = rtts[sj]; sj = sj - 1 } }
      }
      rtts[sj+1] = key
      si = si + 1
  }
  var p50: i64 = 0 - 1
  if nrtt > 0 { p50 = rtts[nrtt / 2] }
268
  // ---- p50 dec fps (median = the robust received-fps floor; MIN would false-RED on the by-design
  // F604 slow-start ramp or one probe-host contention sample -- measured live 2026-07-20) ----
  var di: i64 = 1
  while di < ndec {
      let dkey: i64 = decs[di]
      var dj: i64 = di - 1
      var dmsc: i64 = 1
      while dmsc == 1 {
          if dj < 0 { dmsc = 0 } else { if decs[dj] <= dkey { dmsc = 0 } else { decs[dj+1] = decs[dj]; dj = dj - 1 } }
      }
      decs[dj+1] = dkey
      di = di + 1
  }
  var p50dec: i64 = 0 - 1
  if ndec > 0 { p50dec = decs[ndec / 2] }
284
  var droppct: i64 = 0 - 1
  if rxsum + dksum > 0 { droppct = (dksum * 100) / (rxsum + dksum) }
287
  // ---- verdict ----
  var verdict: i64 = 0
  let why: *u8 = sys_mmap(1024)
  var wo: i64 = 0
  if nb < minb { verdict = 2 } else {
      if p50dec >= 0 { if p50dec < decmin {
          verdict = 1
          wo = vg_s(why, wo, " p50_dec_fps=" as *u8); wo = vg_n(why, wo, p50dec); wo = vg_s(why, wo, "<" as *u8); wo = vg_n(why, wo, decmin)
      } }
      if p50 >= 0 { if p50 > rttmax {
          verdict = 1
          wo = vg_s(why, wo, " p50_rtt_ms=" as *u8); wo = vg_n(why, wo, p50); wo = vg_s(why, wo, ">" as *u8); wo = vg_n(why, wo, rttmax)
      } }
      if droppct >= 0 { if droppct > dropmax {
          verdict = 1
          wo = vg_s(why, wo, " drop_pct=" as *u8); wo = vg_n(why, wo, droppct); wo = vg_s(why, wo, ">" as *u8); wo = vg_n(why, wo, dropmax)
      } }
      if nstarv * 100 > nb * starvmax {
          verdict = 1
          wo = vg_s(why, wo, " decode-starvation=" as *u8); wo = vg_n(why, wo, nstarv); wo = vg_s(why, wo, "/" as *u8); wo = vg_n(why, wo, nb)
      }
  }
  why[wo] = 0 as u8
311
  // ---- build report ----
  let ob: *u8 = sys_mmap(16384)
  var o: i64 = 0
  o = vg_s(ob, o, "nx_vqoe_gate epoch=" as *u8); o = vg_n(ob, o, now)
  o = vg_s(ob, o, "\nevidence=" as *u8); o = vg_s(ob, o, jsonl)
  o = vg_s(ob, o, " file_bytes=" as *u8); o = vg_n(ob, o, fsz)
  o = vg_s(ob, o, " window_sec=" as *u8); o = vg_n(ob, o, window)
  o = vg_s(ob, o, " beacons=" as *u8); o = vg_n(ob, o, nb)
  o = vg_s(ob, o, " probe_rows_excluded=" as *u8); o = vg_n(ob, o, nprobe)
  o = vg_s(ob, o, " newest_age_s=" as *u8)
  if newest > 0 { o = vg_n(ob, o, now - newest) } else { o = vg_n(ob, o, 0 - 1) }
  o = vg_s(ob, o, "\np50_dec_fps=" as *u8)
  if p50dec >= 0 { o = vg_n(ob, o, p50dec) } else { o = vg_s(ob, o, "na" as *u8) }
  o = vg_s(ob, o, " worst_dec_fps=" as *u8)
  if worstdec < 99999 { o = vg_n(ob, o, worstdec) } else { o = vg_s(ob, o, "na" as *u8) }
  o = vg_s(ob, o, " p50_rtt_ms=" as *u8)
  if p50 >= 0 { o = vg_n(ob, o, p50) } else { o = vg_s(ob, o, "na" as *u8) }
  o = vg_s(ob, o, " drop_pct=" as *u8)
  if droppct >= 0 { o = vg_n(ob, o, droppct) } else { o = vg_s(ob, o, "na" as *u8) }
  o = vg_s(ob, o, " (dk=" as *u8); o = vg_n(ob, o, dksum)
  o = vg_s(ob, o, " rx=" as *u8); o = vg_n(ob, o, rxsum); o = vg_s(ob, o, ")" as *u8)
  o = vg_s(ob, o, " starv=" as *u8); o = vg_n(ob, o, nstarv); o = vg_s(ob, o, "/" as *u8); o = vg_n(ob, o, nb)
  o = vg_s(ob, o, "\nthresholds dec_fps_min=" as *u8); o = vg_n(ob, o, decmin)
  o = vg_s(ob, o, " rtt_ms_max=" as *u8); o = vg_n(ob, o, rttmax)
  o = vg_s(ob, o, " drop_pct_max=" as *u8); o = vg_n(ob, o, dropmax)
  o = vg_s(ob, o, " min_beacons=" as *u8); o = vg_n(ob, o, minb)
  o = vg_s(ob, o, " src=" as *u8); o = vg_s(ob, o, confp)
  o = vg_s(ob, o, "\nenvelope tail_bytes=" as *u8); o = vg_n(ob, o, tailb)
  o = vg_s(ob, o, " rtt_cap=" as *u8); o = vg_n(ob, o, VG_RTT_CAP)
  o = vg_s(ob, o, " overcap_dropped=" as *u8); o = vg_n(ob, o, overcap)
  o = vg_s(ob, o, " (DECLARED, never silent)\n" as *u8)
  if verdict == 0 {
      o = vg_s(ob, o, "VERDICT=GREEN received-room QoE within thresholds\n" as *u8)
  }
  if verdict == 2 {
      o = vg_s(ob, o, "VERDICT=AMBER insufficient-fresh-evidence beacons=" as *u8); o = vg_n(ob, o, nb)
      o = vg_s(ob, o, "<" as *u8); o = vg_n(ob, o, minb)
      o = vg_s(ob, o, " -- video ship UNPROVEN (not blocked): generate receiver evidence via the headless room probe or a real family call\n" as *u8)
  }
  if verdict == 1 {
      o = vg_s(ob, o, "VERDICT=RED" as *u8); o = vg_s(ob, o, why)
      o = vg_s(ob, o, " -- received-room truth FAILED; do NOT ship video changes (check relay load, client pacing, codec regression; re-run after fix)\n" as *u8)
  }
  ob[o] = 0 as u8
356
  // ---- publish atomically (tmp + rename; a reader never sees a torn verdict) ----
  let tmpp: *u8 = sys_mmap(1024)
  var tp2: i64 = vg_s(tmpp, 0, outp)
  tp2 = vg_s(tmpp, tp2, ".new" as *u8)
  tmpp[tp2] = 0 as u8
  let ofd: i64 = sys_openat_wr(tmpp, 0x1a4)
  if ofd >= 0 {
      sys_write(ofd, ob, o)
      sys_close(ofd)
      sys_renameat(tmpp, outp)
  }
  vg_puts(ob)
  if verdict == 1 { sys_exit(1); return 1 }
  if verdict == 2 { sys_exit(2); return 2 }
  sys_exit(0)
  return 0
373}
nx_vqoe_gate.nx source

`nx_vqoe_gate.nx` source
