`nx_netobs.nx` source

↩ module page · 1206 lines · 55636 B
1// nx_netobs.nx -- SOVEREIGN NET/HOST OBSERVABILITY + CLOSED-LOOP BENCH (operator 2026-07-20: "make sure
2// our networking and hosting is state of the art ... sovereign version of the Prometheus/Telegraf/
3// Grafana/k6/analyzer stack, largest rock down"). One drift-immune organ (imports ONLY nx_syscalls):
4//   (no-arg|beat)  clock-beat: 2s-window host counters (/proc: cpu,mem,load,net,tcp-snmp) + per-service
5//                  loopback latency probes (manifest-driven) -> METRIC rows (ledger, rotated) -> page.
6//   burst <n> <ms> high-frequency sampler (operator's sub-second ask, bounded envelope) -> same series.
7//   page           emit sites/nishifamily/netobs.html: CORRELATIVE aligned-timeline SVG rows (Grafana role).
8//   bench <conc> <total> <port> <path>  sovereign loadgen (k6 role): fork workers, loopback-ONLY by
9//                  construction, p50/p90/p95/p99 percentiles + rps -> ledger + netobs_bench.log.
10//   verdict        analyzer (closed-loop brain): thresholds from knowledge/registry/netobs_thresholds.tsv
11//                  (rule 11: NO magic numbers in code) vs latest ledger samples -> netobs_gate.log
12//                  VERDICT=GREEN|RED (deploy_ready/cron_watch evidence-pointer format).
13//   selftest       KAT teeth (cpu-parse bounds, percentile KAT, ledger roundtrip, probe).
14// Ledger: knowledge/status/netobs_metrics.log, METRIC line protocol (contract-compatible with
15// nx_metric_ledger; ADDITIVE ts=<epoch> field). Rotation at 3MiB -> .1 (bounded, declared).
16// Rule 26: ZERO hardware writes -- /proc,/sys READS only; tuning half emits RECOMMENDATIONS, never sysctl.
17// Envelopes DECLARED IN OUTPUT (scale law): beat=2s-window, probes<=16, probe-timeout=5s, page=12h/720pt,
18// bench conc<=64 total<=100000 loopback-only, burst n<=600 ms>=200, ledger-rot=3MiB.
19// CONFIG = SOVEREIGN SEG-STORE PLANES (no-tsv law, operator emphatic 2026-06-21/07-20): probes + thresholds
20// live in knowledge/store/netobs-probes- / netobs-thresholds- (read via sts_load = the canonical store-seed
21// codec); `seedcfg` migrates the retired .tsv in one-time; a .tsv read remains ONLY as a zero-downtime fallback.
22// license_tier: ORIGINAL   expect_exit: 0
23import "nx_sovjson_lib.nx"
24import "nx_connect.nx"   // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
25import "nx_syscalls.nx"
26import "nx_store_seed_lib.nx"
27const NO_MAGIC_65536: i64 = 65536
28const NO_MAGIC_65535: i64 = 65535
29const NO_MAGIC_1024: i64 = 1024
30const NO_MAGIC_2000: i64 = 2000
31const NO_MAGIC_131072: i64 = 131072
32const NO_MAGIC_1048576: i64 = 1048576
33const NO_MAGIC_65000: i64 = 65000
34const NO_MAGIC_2097152: i64 = 2097152
35const NO_MAGIC_2097151: i64 = 2097151
36const NO_MAGIC_1000000000: i64 = 1000000000
37const NO_MAGIC_50000: i64 = 50000
38const NO_MAGIC_4096: i64 = 4096
39const NO_MAGIC_7777: i64 = 7777
40const NO_MAGIC_1111: i64 = 1111
41const NO_MAGIC_8888: i64 = 8888
42const NO_MAGIC_2222: i64 = 2222
43const NO_MAGIC_4095: i64 = 4095
44const NO_MAGIC_8443: i64 = 8443
45const NO_MAGIC_18090: i64 = 18090
46
47const NO_CAP: i64 = 4194304
48const NO_ROT: i64 = 3145728
49const NO_WIN: i64 = 43200
50const NO_PTS: i64 = 720
51const NO_PROBES_MAX: i64 = 16
52const NO_BENCH_CONC_MAX: i64 = 64
53const NO_BENCH_TOTAL_MAX: i64 = 100000
54const NO_BURST_N_MAX: i64 = 600
55const NO_BURST_MS_MIN: i64 = 200
56const NO_DRAIN_CAP: i64 = 262144
57const NO_TMO_S: i64 = 5
58
59func no_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
60func no_w(s: *u8) -> i64 { let n: i64 = no_len(s); sys_write(1, s, n); return 0 }
61func no_cat(d: *u8, o: i64, s: *u8) -> i64 { return sj_cat(d, o, s) }
62func no_catn(d: *u8, o: i64, v: i64) -> i64 { return sj_catn(d, o, v) }
63func no_wn(v: i64) -> i64 { let b: *u8 = sys_mmap(32); let o: i64 = no_catn(b, 0, v); sys_write(1, b, o); return 0 }
64func no_eq(a: *u8, b: *u8) -> i64 { var i: i64 = 0; var g: i64 = 1; var r: i64 = 1; while g == 1 { if a[i] != b[i] { r = 0; g = 0 } else { if a[i] == (0 as u8) { g = 0 } else { i = i + 1 } } } return r }
65func no_atoi(s: *u8) -> i64 { var v: i64 = 0; var i: i64 = 0; while s[i] != (0 as u8) { let c: i64 = s[i] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 } return v }
66func no_read(path: *u8, buf: *u8, cap: i64) -> i64 {
  let fd: i64 = sys_openat_rd(path)
  if fd < 0 { return 0 - 1 }
  var n: i64 = 0
  var go: i64 = 1
  while go == 1 { let dst: *u8 = ((buf as i64) + n) as *u8; let r: i64 = sys_read(fd, dst, cap - n); if r <= 0 { go = 0 } else { n = n + r } if n >= cap { go = 0 } }
  sys_close(fd)
  return n
74}
75// no-tsv config read: sovereign seg-store plane first (sts_load), .tsv fallback for zero-downtime transition.
76func no_load_cfg(plane: *u8, tsv: *u8, buf: *u8, cap: i64) -> i64 {
  let pn: i64 = sts_load(plane, buf, cap)
  if pn > 0 { return pn }
  return no_read(tsv, buf, cap)
80}
81func no_wfd(fd: i64, buf: *u8, n: i64) -> i64 { var w: i64 = 0; while w < n { let src: *u8 = ((buf as i64) + w) as *u8; let r: i64 = sys_write(fd, src, n - w); if r <= 0 { w = n } else { w = w + r } } return 0 }
82// atomic full-file write: <path>.nxw tmp + fsync + rename (a reader never sees a torn file)
83func no_atomic(path: *u8, buf: *u8, n: i64) -> i64 {
  let tp: *u8 = sys_mmap(512)
  var o: i64 = no_cat(tp, 0, path)
  o = no_cat(tp, o, ".nxw" as *u8)
  tp[o] = 0 as u8
  let fd: i64 = sys_openat_wr(tp, 420)
  if fd < 0 { return 0 - 1 }
  no_wfd(fd, buf, n)
  sys_fsync(fd)
  sys_close(fd)
  sys_renameat(tp, path)
  return 0
95}
96func no_append(path: *u8, buf: *u8, n: i64) -> i64 {
  var fd: i64 = sys_openat_append(path, 420)
  if fd < 0 { let cf: i64 = sys_openat_wr(path, 420); if cf >= 0 { sys_close(cf) } fd = sys_openat_append(path, 420) }
  if fd < 0 { return 0 - 1 }
  no_wfd(fd, buf, n)
  sys_close(fd)
  return 0
103}
104func no_fsize(path: *u8) -> i64 {
  let stb: *u8 = sys_mmap(160)
  let r: i64 = sys_fstatat(path, stb)
  if r != 0 { return 0 - 1 }
  let sp: *i64 = stb as *i64
  return sp[6]
110}
111// parse the first unsigned decimal AT OR AFTER `marker` (skipping spaces/tabs between). -1 if absent.
112func no_num_after(buf: *u8, n: i64, marker: *u8) -> i64 {
  let ml: i64 = no_len(marker)
  if ml == 0 { return 0 - 1 }
  var i: i64 = 0
  while i + ml <= n {
      var j: i64 = 0
      var ok: i64 = 1
      while j < ml { if buf[i+j] != marker[j] { ok = 0; j = ml } else { j = j + 1 } }
      if ok == 1 {
          var p: i64 = i + ml
          var sk: i64 = 1
          while sk == 1 { if p < n { let sc: i64 = buf[p] as i64; if sc == 32 { p = p + 1 } else { if sc == 9 { p = p + 1 } else { sk = 0 } } } else { sk = 0 } }
          var v: i64 = 0
          var got: i64 = 0
          var go: i64 = 1
          while go == 1 { if p >= n { go = 0 } else { let c: i64 = buf[p] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48); got = 1; p = p + 1 } else { go = 0 } } else { go = 0 } } }
          if got == 1 { return v }
      }
      i = i + 1
  }
  return 0 - 1
133}
134func no_line_end(q: *u8, i: i64, n: i64) -> i64 { var e: i64 = i; var s: i64 = 1; while s == 1 { if e >= n { s = 0 } else { if q[e] == (10 as u8) { s = 0 } else { e = e + 1 } } } return e }
135// TSV column c span within [ls,le) -> out[0]=start out[1]=end; 1 ok / 0 missing
136func no_col(q: *u8, ls: i64, le: i64, c: i64, out: *i64) -> i64 {
  var col: i64 = 0
  var p: i64 = ls
  while col < c {
      var s: i64 = 1
      while s == 1 { if p >= le { return 0 } if q[p] == (9 as u8) { s = 0 } else { p = p + 1 } }
      p = p + 1
      col = col + 1
  }
  var e: i64 = p
  var s2: i64 = 1
  while s2 == 1 { if e >= le { s2 = 0 } else { if q[e] == (9 as u8) { s2 = 0 } else { e = e + 1 } } }
  out[0] = p
  out[1] = e
  return 1
151}
152func no_cstr(q: *u8, s: i64, e: i64, dst: *u8) -> i64 { var i: i64 = 0; while s + i < e { dst[i] = q[s+i]; i = i + 1 } dst[i] = 0 as u8; return i }
153func no_span_atoi(q: *u8, s: i64, e: i64) -> i64 { var v: i64 = 0; var i: i64 = s; while i < e { let c: i64 = q[i] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 } return v }
154// nth whitespace-separated token AFTER position p0 on the line ending at le; returns value or -1
155func no_tokn(q: *u8, p0: i64, le: i64, nth: i64) -> i64 {
  var p: i64 = p0
  var t: i64 = 0
  while t < nth {
      var s1: i64 = 1
      while s1 == 1 { if p >= le { return 0 - 1 } if q[p] == (32 as u8) { p = p + 1 } else { s1 = 0 } }
      var s2: i64 = 1
      while s2 == 1 { if p >= le { s2 = 0 } else { if q[p] == (32 as u8) { s2 = 0 } else { p = p + 1 } } }
      t = t + 1
  }
  var s3: i64 = 1
  while s3 == 1 { if p >= le { return 0 - 1 } if q[p] == (32 as u8) { p = p + 1 } else { s3 = 0 } }
  var v: i64 = 0
  var got: i64 = 0
  var s4: i64 = 1
  while s4 == 1 { if p >= le { s4 = 0 } else { let c: i64 = q[p] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48); got = 1; p = p + 1 } else { s4 = 0 } } else { s4 = 0 } } }
  if got == 0 { return 0 - 1 }
  return v
173}
174
175// ---------------- /proc samplers (READ-only; rule 26 holds by construction) ----------------
176// out[0]=busy jiffies, out[1]=total jiffies. 1 ok / 0 fail.
177func no_cpu(out: *i64) -> i64 {
  let buf: *u8 = sys_mmap(NO_MAGIC_65536)
  let n: i64 = no_read("/proc/stat" as *u8, buf, NO_MAGIC_65535)
  if n <= 4 { return 0 }
  let le: i64 = no_line_end(buf, 0, n)
  var p: i64 = 3
  var total: i64 = 0
  var idle: i64 = 0
  var iow: i64 = 0
  var k: i64 = 0
  var go: i64 = 1
  while go == 1 {
      var s1: i64 = 1
      while s1 == 1 { if p >= le { s1 = 0 } else { if buf[p] == (32 as u8) { p = p + 1 } else { s1 = 0 } } }
      if p >= le { go = 0 } else {
          var v: i64 = 0
          var got: i64 = 0
          var s2: i64 = 1
          while s2 == 1 { if p >= le { s2 = 0 } else { let c: i64 = buf[p] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48); got = 1; p = p + 1 } else { s2 = 0 } } else { s2 = 0 } } }
          if got == 0 { go = 0 } else { k = k + 1; total = total + v; if k == 4 { idle = v } if k == 5 { iow = v } }
      }
  }
  if k < 5 { return 0 }
  out[0] = total - idle - iow
  out[1] = total
  return 1
203}
204func no_mem_avail_mb() -> i64 {
  let buf: *u8 = sys_mmap(NO_MAGIC_65536)
  let n: i64 = no_read("/proc/meminfo" as *u8, buf, NO_MAGIC_65535)
  if n <= 0 { return 0 - 1 }
  let kb: i64 = no_num_after(buf, n, "MemAvailable:" as *u8)
  if kb < 0 { return 0 - 1 }
  return kb / NO_MAGIC_1024
211}
212func no_load_milli() -> i64 {
  let buf: *u8 = sys_mmap(256)
  let n: i64 = no_read("/proc/loadavg" as *u8, buf, 255)
  if n <= 0 { return 0 - 1 }
  var p: i64 = 0
  var ip: i64 = 0
  var s1: i64 = 1
  while s1 == 1 { if p >= n { s1 = 0 } else { let c: i64 = buf[p] as i64; if c >= 48 { if c <= 57 { ip = ip * 10 + (c - 48); p = p + 1 } else { s1 = 0 } } else { s1 = 0 } } }
  var fr: i64 = 0
  if p < n { if buf[p] == (46 as u8) { p = p + 1; var d: i64 = 0; while d < 2 { if p < n { let c2: i64 = buf[p] as i64; if c2 >= 48 { if c2 <= 57 { fr = fr * 10 + (c2 - 48); p = p + 1 } } } d = d + 1 } } }
  return ip * 1000 + fr * 10
223}
224// out[0]=rx bytes sum, out[1]=tx bytes sum across non-lo ifaces. 1 ok / 0 fail.
225func no_netdev(out: *i64) -> i64 {
  let buf: *u8 = sys_mmap(NO_MAGIC_65536)
  let n: i64 = no_read("/proc/net/dev" as *u8, buf, NO_MAGIC_65535)
  if n <= 0 { return 0 }
  var rx: i64 = 0
  var tx: i64 = 0
  var i: i64 = 0
  var lines: i64 = 0
  while i < n {
      let le: i64 = no_line_end(buf, i, n)
      if lines >= 2 {
          var cp: i64 = 0 - 1
          var q: i64 = i
          while q < le { if buf[q] == (58 as u8) { cp = q; q = le } else { q = q + 1 } }
          if cp > 0 {
              var ns: i64 = i
              var s1: i64 = 1
              while s1 == 1 { if ns < cp { if buf[ns] == (32 as u8) { ns = ns + 1 } else { s1 = 0 } } else { s1 = 0 } }
              var islo: i64 = 0
              if cp - ns == 2 { if buf[ns] == (108 as u8) { if buf[ns+1] == (111 as u8) { islo = 1 } } }
              if islo == 0 {
                  let r1: i64 = no_tokn(buf, cp + 1, le, 0)
                  let t1: i64 = no_tokn(buf, cp + 1, le, 8)
                  if r1 >= 0 { rx = rx + r1 }
                  if t1 >= 0 { tx = tx + t1 }
              }
          }
      }
      lines = lines + 1
      i = le + 1
  }
  out[0] = rx
  out[1] = tx
  return 1
259}
260// header-indexed /proc/net/snmp Tcp: parse. out[0]=CurrEstab out[1]=OutSegs out[2]=RetransSegs. 1 ok/0 fail.
261func no_snmp(out: *i64) -> i64 {
  let buf: *u8 = sys_mmap(NO_MAGIC_65536)
  let n: i64 = no_read("/proc/net/snmp" as *u8, buf, NO_MAGIC_65535)
  if n <= 0 { return 0 }
  var h0: i64 = 0 - 1
  var h1: i64 = 0 - 1
  var i: i64 = 0
  while i < n {
      let le: i64 = no_line_end(buf, i, n)
      if le - i > 4 { if buf[i] == (84 as u8) { if buf[i+1] == (99 as u8) { if buf[i+2] == (112 as u8) { if buf[i+3] == (58 as u8) { if h0 < 0 { h0 = i } else { if h1 < 0 { h1 = i } } } } } } }
      i = le + 1
  }
  if h1 < 0 { return 0 }
  let h0e: i64 = no_line_end(buf, h0, n)
  let h1e: i64 = no_line_end(buf, h1, n)
  // find label token indices on the header line
  var iCE: i64 = 0 - 1
  var iOS: i64 = 0 - 1
  var iRS: i64 = 0 - 1
  var p: i64 = h0 + 4
  var tok: i64 = 0
  var go: i64 = 1
  while go == 1 {
      var s1: i64 = 1
      while s1 == 1 { if p >= h0e { s1 = 0 } else { if buf[p] == (32 as u8) { p = p + 1 } else { s1 = 0 } } }
      if p >= h0e { go = 0 } else {
          var e: i64 = p
          var s2: i64 = 1
          while s2 == 1 { if e >= h0e { s2 = 0 } else { if buf[e] == (32 as u8) { s2 = 0 } else { e = e + 1 } } }
          if e - p == 9 { if buf[p] == (67 as u8) { if buf[p+4] == (69 as u8) { iCE = tok } } }
          if e - p == 7 { if buf[p] == (79 as u8) { if buf[p+3] == (83 as u8) { iOS = tok } } }
          if e - p == 11 { if buf[p] == (82 as u8) { if buf[p+7] == (83 as u8) { iRS = tok } } }
          tok = tok + 1
          p = e
      }
  }
  if iCE < 0 { return 0 }
  if iOS < 0 { return 0 }
  if iRS < 0 { return 0 }
  out[0] = no_tokn(buf, h1 + 4, h1e, iCE)
  out[1] = no_tokn(buf, h1 + 4, h1e, iOS)
  out[2] = no_tokn(buf, h1 + 4, h1e, iRS)
  return 1
304}
305
306// ---------------- loopback probe (latency eye) ----------------
307// mode 0=connect-only, 1=GET. out[0]=connect_us out[1]=ttfb_us(-1 n/a) out[2]=ok. returns ok.
308// scr = CALLER-OWNED scratch >= 131072 bytes (sa/rq/rb live inside) -- NO per-call mmap, so a
309// 100k-request bench worker stays flat-RSS (the sys_read_file OOM-class lesson, rule 21).
310func no_probe(port: i64, path: *u8, mode: i64, out: *i64, scr: *u8) -> i64 {
  out[0] = 0 - 1
  out[1] = 0 - 1
  out[2] = 0
  let fd: i64 = sys_socket(2, 1, 0)
  if fd < 0 { return 0 }
  sys_set_socket_timeout(fd, NO_TMO_S)
  let sa: *u8 = scr
  sa[0] = 2 as u8
  sa[1] = 0 as u8
  sa[2] = ((port >> 8) & 0xff) as u8
  sa[3] = (port & 0xff) as u8
  sa[4] = 127 as u8
  sa[5] = 0 as u8
  sa[6] = 0 as u8
  sa[7] = 1 as u8
  var zi: i64 = 8
  while zi < 16 { sa[zi] = 0 as u8; zi = zi + 1 }
  let t0: i64 = sys_clock_now_us()
  let cr: i64 = nx_connect_bounded(fd, sa, 16, NX_CONN_DEFAULT_MS)
  let t1: i64 = sys_clock_now_us()
  if cr < 0 { sys_close(fd); return 0 }
  out[0] = t1 - t0
  if mode == 0 { out[2] = 1; sys_close(fd); return 1 }
  let rq: *u8 = ((scr as i64) + 16) as *u8
  var o: i64 = no_cat(rq, 0, "GET " as *u8)
  o = no_cat(rq, o, path)
  o = no_cat(rq, o, " HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n" as *u8)
  no_wfd(fd, rq, o)
  let rb: *u8 = ((scr as i64) + NO_MAGIC_1024) as *u8
  let r0: i64 = sys_read(fd, rb, NO_MAGIC_65536)
  let t2: i64 = sys_clock_now_us()
  if r0 <= 0 { sys_close(fd); return 0 }
  out[1] = t2 - t0
  out[2] = 1
  var tot: i64 = r0
  var r: i64 = r0
  while r > 0 { if tot >= NO_DRAIN_CAP { r = 0 } else { r = sys_read(fd, rb, NO_MAGIC_65536); if r > 0 { tot = tot + r } } }
  sys_close(fd)
  return 1
350}
351// one full request (connect->drain) latency in us, or -1. scr layout: [0..32)=out, [32..) probe scratch.
352func no_req(port: i64, path: *u8, scr: *u8) -> i64 {
  let out: *i64 = scr as *i64
  let ps: *u8 = ((scr as i64) + 32) as *u8
  let t0: i64 = sys_clock_now_us()
  let ok: i64 = no_probe(port, path, 1, out, ps)
  let t1: i64 = sys_clock_now_us()
  if ok == 0 { return 0 - 1 }
  return t1 - t0
360}
361
362// ---------------- ledger ----------------
363func no_mrow(lb: *u8, o: i64, name: *u8, tu: i64, ts: i64, val: i64, unit: *u8) -> i64 {
  var p: i64 = no_cat(lb, o, "METRIC " as *u8)
  p = no_cat(lb, p, name)
  p = no_cat(lb, p, " t_us=" as *u8)
  p = no_catn(lb, p, tu)
  p = no_cat(lb, p, " ts=" as *u8)
  p = no_catn(lb, p, ts)
  p = no_cat(lb, p, " value=" as *u8)
  p = no_catn(lb, p, val)
  p = no_cat(lb, p, " unit=" as *u8)
  p = no_cat(lb, p, unit)
  p = no_cat(lb, p, " src=nx_netobs\n" as *u8)
  return p
376}
377// last METRIC row for `name`: out[0]=ts out[1]=value; 1 found / 0 not
378func no_last(buf: *u8, n: i64, name: *u8, out: *i64) -> i64 {
  let nb: *u8 = sys_mmap(160)
  var o: i64 = no_cat(nb, 0, "METRIC " as *u8)
  o = no_cat(nb, o, name)
  o = no_cat(nb, o, " " as *u8)
  nb[o] = 0 as u8
  let nl: i64 = o
  var best: i64 = 0 - 1
  var i: i64 = 0
  while i < n {
      let le: i64 = no_line_end(buf, i, n)
      if le - i > nl {
          var j: i64 = 0
          var ok: i64 = 1
          while j < nl { if buf[i+j] != nb[j] { ok = 0; j = nl } else { j = j + 1 } }
          if ok == 1 { best = i }
      }
      i = le + 1
  }
  if best < 0 { return 0 }
  let ble: i64 = no_line_end(buf, best, n)
  let seg: *u8 = ((buf as i64) + best) as *u8
  let sn: i64 = ble - best
  out[0] = no_num_after(seg, sn, " ts=" as *u8)
  out[1] = no_num_after(seg, sn, " value=" as *u8)
  return 1
404}
405// collect series points for `name` with ts>=ts0: two-pass (count, then store last<=maxp). returns count stored.
406func no_series(buf: *u8, n: i64, name: *u8, ts0: i64, xs: *i64, ys: *i64, maxp: i64) -> i64 {
  let nb: *u8 = sys_mmap(160)
  var o: i64 = no_cat(nb, 0, "METRIC " as *u8)
  o = no_cat(nb, o, name)
  o = no_cat(nb, o, " " as *u8)
  nb[o] = 0 as u8
  let nl: i64 = o
  var total: i64 = 0
  var pass: i64 = 0
  var stored: i64 = 0
  while pass < 2 {
      var seen: i64 = 0
      var i: i64 = 0
      while i < n {
          let le: i64 = no_line_end(buf, i, n)
          if le - i > nl {
              var j: i64 = 0
              var ok: i64 = 1
              while j < nl { if buf[i+j] != nb[j] { ok = 0; j = nl } else { j = j + 1 } }
              if ok == 1 {
                  let seg: *u8 = ((buf as i64) + i) as *u8
                  let sn: i64 = le - i
                  let ts: i64 = no_num_after(seg, sn, " ts=" as *u8)
                  if ts >= ts0 {
                      if pass == 0 { total = total + 1 } else {
                          var skip: i64 = 0
                          if total > maxp { skip = total - maxp }
                          if seen >= skip { if stored < maxp { let vv: i64 = no_num_after(seg, sn, " value=" as *u8); xs[stored] = ts; ys[stored] = vv; stored = stored + 1 } }
                          seen = seen + 1
                      }
                  }
              }
          }
          i = le + 1
      }
      pass = pass + 1
  }
  return stored
444}
445
446// ---------------- sort + percentiles (bench brain) ----------------
447func no_sift(a: *i64, n: i64, i0: i64) -> i64 {
  var i: i64 = i0
  var g: i64 = 1
  while g == 1 {
      var c: i64 = 2 * i + 1
      if c >= n { g = 0 } else {
          let c2: i64 = c + 1
          if c2 < n { if a[c2] > a[c] { c = c2 } }
          if a[c] > a[i] { let t: i64 = a[i]; a[i] = a[c]; a[c] = t; i = c } else { g = 0 }
      }
  }
  return 0
459}
460func no_sort(a: *i64, n: i64) -> i64 {
  if n < 2 { return 0 }
  var i: i64 = n / 2 - 1
  while i >= 0 { no_sift(a, n, i); i = i - 1 }
  var e: i64 = n - 1
  while e > 0 { let t: i64 = a[0]; a[0] = a[e]; a[e] = t; no_sift(a, e, 0); e = e - 1 }
  return 0
467}
468func no_pct(a: *i64, n: i64, p: i64) -> i64 { if n <= 0 { return 0 - 1 } let ix: i64 = ((n - 1) * p) / 100; return a[ix] }
469
470// ---------------- beat ----------------
471func no_beat(do_page: i64) -> i64 {
  let now: i64 = sys_now_realtime_sec()
  let a2: *i64 = sys_mmap(32) as *i64
  let b2: *i64 = sys_mmap(32) as *i64
  let na: *i64 = sys_mmap(32) as *i64
  let nbv: *i64 = sys_mmap(32) as *i64
  let s3a: *i64 = sys_mmap(32) as *i64
  let s3b: *i64 = sys_mmap(32) as *i64
  let cok: i64 = no_cpu(a2)
  let nok: i64 = no_netdev(na)
  let sok: i64 = no_snmp(s3a)
  let tA: i64 = sys_clock_now_us()
  sys_sleep_ms(NO_MAGIC_2000)
  let cok2: i64 = no_cpu(b2)
  let nok2: i64 = no_netdev(nbv)
  let sok2: i64 = no_snmp(s3b)
  let tB: i64 = sys_clock_now_us()
  var dt_ms: i64 = (tB - tA) / 1000
  if dt_ms < 1 { dt_ms = 1 }
  let lb: *u8 = sys_mmap(NO_MAGIC_65536)
  var o: i64 = 0
  let tu: i64 = sys_now_us()
  if cok == 1 { if cok2 == 1 {
      let dtot: i64 = b2[1] - a2[1]
      let dbusy: i64 = b2[0] - a2[0]
      var permil: i64 = 0
      if dtot > 0 { permil = (dbusy * 1000) / dtot }
      if permil < 0 { permil = 0 }
      if permil > 1000 { permil = 1000 }
      o = no_mrow(lb, o, "netobs_cpu_busy_permil" as *u8, tu, now, permil, "permil" as *u8)
  } }
  let mem: i64 = no_mem_avail_mb()
  if mem >= 0 { o = no_mrow(lb, o, "netobs_mem_avail_mb" as *u8, tu, now, mem, "MB" as *u8) }
  let lm: i64 = no_load_milli()
  if lm >= 0 { o = no_mrow(lb, o, "netobs_load1_milli" as *u8, tu, now, lm, "milliload" as *u8) }
  if nok == 1 { if nok2 == 1 {
      let drx: i64 = nbv[0] - na[0]
      let dtx: i64 = nbv[1] - na[1]
      var rxk: i64 = (drx * 8) / dt_ms
      var txk: i64 = (dtx * 8) / dt_ms
      if rxk < 0 { rxk = 0 }
      if txk < 0 { txk = 0 }
      o = no_mrow(lb, o, "netobs_net_rx_kbps" as *u8, tu, now, rxk, "kbps" as *u8)
      o = no_mrow(lb, o, "netobs_net_tx_kbps" as *u8, tu, now, txk, "kbps" as *u8)
  } }
  if sok == 1 { if sok2 == 1 {
      var drs: i64 = s3b[2] - s3a[2]
      if drs < 0 { drs = 0 }
      var dos: i64 = s3b[1] - s3a[1]
      if dos < 0 { dos = 0 }
      o = no_mrow(lb, o, "netobs_tcp_retrans_win" as *u8, tu, now, drs, "segs2s" as *u8)
      o = no_mrow(lb, o, "netobs_tcp_outsegs_win" as *u8, tu, now, dos, "segs2s" as *u8)
      o = no_mrow(lb, o, "netobs_tcp_estab" as *u8, tu, now, s3b[0], "conns" as *u8)
  } }
  // probes (manifest-driven, envelope <=16)
  let man: *u8 = sys_mmap(NO_MAGIC_65536)
  let mn: i64 = no_load_cfg("knowledge/store/netobs-probes-" as *u8, "knowledge/registry/netobs_probes.tsv" as *u8, man, NO_MAGIC_65535)
  var np: i64 = 0
  let c0: *i64 = sys_mmap(16) as *i64
  let c1: *i64 = sys_mmap(16) as *i64
  let c2: *i64 = sys_mmap(16) as *i64
  let c3: *i64 = sys_mmap(16) as *i64
  let pnb: *u8 = sys_mmap(128)
  let mdb: *u8 = sys_mmap(64)
  let mname: *u8 = sys_mmap(128)
  let mpath: *u8 = sys_mmap(256)
  let pr3: *i64 = sys_mmap(32) as *i64
  let pscr: *u8 = sys_mmap(NO_MAGIC_131072)
  var mi: i64 = 0
  while mi < mn {
      let le: i64 = no_line_end(man, mi, mn)
      if le > mi { if (man[mi] as i64) != 35 { if np < NO_PROBES_MAX {
          if no_col(man, mi, le, 0, c0) == 1 { if no_col(man, mi, le, 1, c1) == 1 { if no_col(man, mi, le, 2, c2) == 1 { if no_col(man, mi, le, 3, c3) == 1 {
              no_cstr(man, c0[0], c0[1], mname)
              let port: i64 = no_span_atoi(man, c1[0], c1[1])
              no_cstr(man, c2[0], c2[1], mpath)
              no_cstr(man, c3[0], c3[1], mdb)
              var mode: i64 = 0
              if mdb[0] == (103 as u8) { mode = 1 }
              no_probe(port, mpath, mode, pr3, pscr)
              var po: i64 = no_cat(pnb, 0, "netobs_" as *u8)
              po = no_cat(pnb, po, mname)
              po = no_cat(pnb, po, "_conn_us" as *u8)
              pnb[po] = 0 as u8
              if pr3[0] >= 0 { o = no_mrow(lb, o, pnb, tu, now, pr3[0], "us" as *u8) }
              if mode == 1 { if pr3[1] >= 0 {
                  var po2: i64 = no_cat(pnb, 0, "netobs_" as *u8)
                  po2 = no_cat(pnb, po2, mname)
                  po2 = no_cat(pnb, po2, "_ttfb_us" as *u8)
                  pnb[po2] = 0 as u8
                  o = no_mrow(lb, o, pnb, tu, now, pr3[1], "us" as *u8)
              } }
              var po3: i64 = no_cat(pnb, 0, "netobs_" as *u8)
              po3 = no_cat(pnb, po3, mname)
              po3 = no_cat(pnb, po3, "_ok" as *u8)
              pnb[po3] = 0 as u8
              o = no_mrow(lb, o, pnb, tu, now, pr3[2], "bool" as *u8)
              np = np + 1
          } } } }
      } } }
      mi = le + 1
  }
  // rotation (bounded ledger, declared) then single append
  var lsz: i64 = no_fsize("knowledge/status/netobs_metrics.log" as *u8)
  if lsz > NO_ROT { sys_renameat("knowledge/status/netobs_metrics.log" as *u8, "knowledge/status/netobs_metrics.log.1" as *u8); lsz = 0 }
  no_append("knowledge/status/netobs_metrics.log" as *u8, lb, o)
  // heartbeat (dead-man's-switch food for nx_cron_watch)
  let hb: *u8 = sys_mmap(512)
  var ho: i64 = no_cat(hb, 0, "NETOBS-BEAT ts=" as *u8)
  ho = no_catn(hb, ho, now)
  ho = no_cat(hb, ho, " ledger_bytes=" as *u8)
  ho = no_catn(hb, ho, lsz)
  ho = no_cat(hb, ho, " probes=" as *u8)
  ho = no_catn(hb, ho, np)
  ho = no_cat(hb, ho, " envelope=2s-window,probes-max-16,probe-tmo-5s,rot-3MiB\n" as *u8)
  no_atomic("knowledge/status/netobs_beat.log" as *u8, hb, ho)
  no_w("NETOBS-BEAT ok probes=" as *u8)
  no_wn(np)
  no_w(" bytes=" as *u8)
  no_wn(o)
  no_w(" envelope=2s-window,probes-max-16,probe-tmo-5s,rot-3MiB\n" as *u8)
  if do_page == 1 { no_page() }
  return 0
594}
595
596// ---------------- burst (high-frequency sampler; same series, denser points) ----------------
597func no_burst(n: i64, ms: i64) -> i64 {
  var nn: i64 = n
  var mm: i64 = ms
  if nn > NO_BURST_N_MAX { no_w("NETOBS-BURST REFUSED n>600 (declared envelope)\n" as *u8); sys_exit(3); return 3 }
  if nn < 1 { nn = 1 }
  if mm < NO_BURST_MS_MIN { no_w("NETOBS-BURST REFUSED ms<200 (declared envelope)\n" as *u8); sys_exit(3); return 3 }
  let a2: *i64 = sys_mmap(32) as *i64
  let b2: *i64 = sys_mmap(32) as *i64
  let na: *i64 = sys_mmap(32) as *i64
  let nbv: *i64 = sys_mmap(32) as *i64
  let lb: *u8 = sys_mmap(NO_MAGIC_65536)
  var pc: i64 = no_cpu(a2)
  var pn: i64 = no_netdev(na)
  var pt: i64 = sys_clock_now_us()
  var i: i64 = 0
  while i < nn {
      sys_sleep_ms(mm)
      let cc: i64 = no_cpu(b2)
      let cn: i64 = no_netdev(nbv)
      let ct: i64 = sys_clock_now_us()
      let now: i64 = sys_now_realtime_sec()
      let tu: i64 = sys_now_us()
      var dt_ms: i64 = (ct - pt) / 1000
      if dt_ms < 1 { dt_ms = 1 }
      var o: i64 = 0
      if pc == 1 { if cc == 1 {
          let dtot: i64 = b2[1] - a2[1]
          let dbusy: i64 = b2[0] - a2[0]
          var permil: i64 = 0
          if dtot > 0 { permil = (dbusy * 1000) / dtot }
          if permil < 0 { permil = 0 }
          if permil > 1000 { permil = 1000 }
          o = no_mrow(lb, o, "netobs_cpu_busy_permil" as *u8, tu, now, permil, "permil" as *u8)
      } }
      if pn == 1 { if cn == 1 {
          let drx: i64 = nbv[0] - na[0]
          var rxk: i64 = (drx * 8) / dt_ms
          if rxk < 0 { rxk = 0 }
          o = no_mrow(lb, o, "netobs_net_rx_kbps" as *u8, tu, now, rxk, "kbps" as *u8)
      } }
      if o > 0 { no_append("knowledge/status/netobs_metrics.log" as *u8, lb, o) }
      a2[0] = b2[0]
      a2[1] = b2[1]
      na[0] = nbv[0]
      na[1] = nbv[1]
      pc = cc
      pn = cn
      pt = ct
      i = i + 1
  }
  no_w("NETOBS-BURST done n=" as *u8)
  no_wn(nn)
  no_w(" ms=" as *u8)
  no_wn(mm)
  no_w(" envelope=n-max-600,ms-min-200\n" as *u8)
  return 0
653}
654
655// ---------------- page (correlative aligned-timeline dashboard) ----------------
656func no_svg_series(pg: *u8, po: i64, buf: *u8, n: i64, name: *u8, unit: *u8, ts0: i64, tsN: i64) -> i64 {
  let xs: *i64 = sys_mmap(NO_PTS * 8) as *i64
  let ys: *i64 = sys_mmap(NO_PTS * 8) as *i64
  let cnt: i64 = no_series(buf, n, name, ts0, xs, ys, NO_PTS)
  var o: i64 = no_cat(pg, po, "<div class=row><span class=lab>" as *u8)
  o = no_cat(pg, o, name)
  o = no_cat(pg, o, "</span> <span class=val>last=" as *u8)
  var last: i64 = 0 - 1
  var ymax: i64 = 1
  var ymin: i64 = 0 - 1
  var i: i64 = 0
  while i < cnt { if ys[i] > ymax { ymax = ys[i] } if ymin < 0 { ymin = ys[i] } if ys[i] < ymin { ymin = ys[i] } i = i + 1 }
  if cnt > 0 { let ci: i64 = cnt - 1; last = ys[ci] }
  o = no_catn(pg, o, last)
  o = no_cat(pg, o, " " as *u8)
  o = no_cat(pg, o, unit)
  o = no_cat(pg, o, "</span> min=" as *u8)
  o = no_catn(pg, o, ymin)
  o = no_cat(pg, o, " max=" as *u8)
  o = no_catn(pg, o, ymax)
  o = no_cat(pg, o, " pts=" as *u8)
  o = no_catn(pg, o, cnt)
  o = no_cat(pg, o, "<br><svg width=720 height=48>" as *u8)
  if cnt > 1 {
      o = no_cat(pg, o, "<polyline fill=\"none\" stroke=\"#7aa2f7\" stroke-width=\"1\" points=\"" as *u8)
      var dt: i64 = tsN - ts0
      if dt < 1 { dt = 1 }
      var k: i64 = 0
      while k < cnt {
          let px: i64 = ((xs[k] - ts0) * 718) / dt + 1
          let py: i64 = 46 - ((ys[k] * 44) / ymax)
          o = no_catn(pg, o, px)
          o = no_cat(pg, o, "," as *u8)
          o = no_catn(pg, o, py)
          o = no_cat(pg, o, " " as *u8)
          k = k + 1
      }
      o = no_cat(pg, o, "\"/>" as *u8)
  }
  o = no_cat(pg, o, "</svg></div>\n" as *u8)
  return o
697}
698func no_page() -> i64 {
  let now: i64 = sys_now_realtime_sec()
  let ts0: i64 = now - NO_WIN
  let buf: *u8 = sys_mmap(NO_CAP)
  let n: i64 = no_read("knowledge/status/netobs_metrics.log" as *u8, buf, NO_CAP)
  let pg: *u8 = sys_mmap(NO_MAGIC_1048576)
  var o: i64 = no_cat(pg, 0, "<!doctype html><html><head><meta charset=\"utf-8\"><title>NISHI NETOBS</title><style>body{background:#0b0e14;color:#d7dce2;font-family:monospace;margin:20px}.row{margin:8px 0;border:1px solid #223;padding:6px}svg{background:#10141c;display:block}.lab{color:#7aa2f7;font-weight:bold}.val{color:#9ece6a}pre{color:#8a919e;white-space:pre-wrap;border:1px solid #223;padding:6px}</style></head><body><h2>NISHI NET OBSERVATORY</h2><p>sovereign metric plane -- correlative rows share ONE 12h time axis (x aligned across every row). generated ts=" as *u8)
  o = no_catn(pg, o, now)
  o = no_cat(pg, o, "</p>\n<h3>service latency (loopback probes)</h3>\n" as *u8)
  // probe rows from the manifest so page + collector stay in lockstep
  let man: *u8 = sys_mmap(NO_MAGIC_65536)
  let mn: i64 = no_load_cfg("knowledge/store/netobs-probes-" as *u8, "knowledge/registry/netobs_probes.tsv" as *u8, man, NO_MAGIC_65535)
  let c0: *i64 = sys_mmap(16) as *i64
  let c3: *i64 = sys_mmap(16) as *i64
  let mname: *u8 = sys_mmap(128)
  let mdb: *u8 = sys_mmap(64)
  let pnb: *u8 = sys_mmap(160)
  var np: i64 = 0
  var mi: i64 = 0
  while mi < mn {
      let le: i64 = no_line_end(man, mi, mn)
      if le > mi { if (man[mi] as i64) != 35 { if np < NO_PROBES_MAX {
          if no_col(man, mi, le, 0, c0) == 1 { if no_col(man, mi, le, 3, c3) == 1 {
              no_cstr(man, c0[0], c0[1], mname)
              no_cstr(man, c3[0], c3[1], mdb)
              var po: i64 = no_cat(pnb, 0, "netobs_" as *u8)
              po = no_cat(pnb, po, mname)
              if mdb[0] == (103 as u8) { po = no_cat(pnb, po, "_ttfb_us" as *u8) } else { po = no_cat(pnb, po, "_conn_us" as *u8) }
              pnb[po] = 0 as u8
              o = no_svg_series(pg, o, buf, n, pnb, "us" as *u8, ts0, now)
              np = np + 1
          } }
      } } }
      mi = le + 1
  }
  o = no_cat(pg, o, "<h3>host (cpu / mem / load)</h3>\n" as *u8)
  o = no_svg_series(pg, o, buf, n, "netobs_cpu_busy_permil" as *u8, "permil" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_mem_avail_mb" as *u8, "MB" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_load1_milli" as *u8, "milliload" as *u8, ts0, now)
  o = no_cat(pg, o, "<h3>network (throughput / tcp)</h3>\n" as *u8)
  o = no_svg_series(pg, o, buf, n, "netobs_net_rx_kbps" as *u8, "kbps" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_net_tx_kbps" as *u8, "kbps" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_tcp_retrans_win" as *u8, "segs2s" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_tcp_estab" as *u8, "conns" as *u8, ts0, now)
  o = no_cat(pg, o, "<h3>TLS edge handshake (F799 certloop cure -- warm should stay well under cold)</h3>\n" as *u8)
  o = no_svg_series(pg, o, buf, n, "netobs_tls_hs_cold_ms" as *u8, "ms" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_tls_hs_warm_ms" as *u8, "ms" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_tls_certloop_saved_ms" as *u8, "ms" as *u8, ts0, now)
  o = no_cat(pg, o, "<h3>HTTP keep-alive connection reuse (1 handshake + K GETs vs K handshakes)</h3>\n" as *u8)
  o = no_svg_series(pg, o, buf, n, "netobs_keepalive_batch_ms" as *u8, "ms" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_noreuse_batch_ms" as *u8, "ms" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_keepalive_saved_ms" as *u8, "ms" as *u8, ts0, now)
  o = no_cat(pg, o, "<h3>capacity ramp (Step A -- saturation ladder to the knee; conc where p99 breaches 50ms)</h3>\n" as *u8)
  o = no_svg_series(pg, o, buf, n, "netobs_ramp_max_sustained_conc" as *u8, "conns" as *u8, ts0, now)
  o = no_svg_series(pg, o, buf, n, "netobs_ramp_peak_rps_milli" as *u8, "millirps" as *u8, ts0, now)
  o = no_cat(pg, o, "<h3>verdict (analyzer)</h3>\n<pre>" as *u8)
  let gb: *u8 = sys_mmap(NO_MAGIC_65536)
  let gn: i64 = no_read("knowledge/status/netobs_gate.log" as *u8, gb, NO_MAGIC_65000)
  if gn > 0 { var gi: i64 = 0; while gi < gn { let gc: i64 = gb[gi] as i64; if gc == 60 { pg[o] = 40 as u8 } else { if gc == 62 { pg[o] = 41 as u8 } else { pg[o] = gc as u8 } } o = o + 1; gi = gi + 1 } } else { o = no_cat(pg, o, "no verdict yet -- run: nx_netobs verdict" as *u8) }
  o = no_cat(pg, o, "</pre>\n<h3>bench (latest loadgen run)</h3>\n<pre>" as *u8)
  let bb: *u8 = sys_mmap(NO_MAGIC_65536)
  let bn: i64 = no_read("knowledge/status/netobs_bench.log" as *u8, bb, NO_MAGIC_65000)
  if bn > 0 { var bi: i64 = 0; while bi < bn { let bc: i64 = bb[bi] as i64; if bc == 60 { pg[o] = 40 as u8 } else { if bc == 62 { pg[o] = 41 as u8 } else { pg[o] = bc as u8 } } o = o + 1; bi = bi + 1 } } else { o = no_cat(pg, o, "no bench yet -- run: nx_netobs bench 8 2000 18090 /" as *u8) }
  o = no_cat(pg, o, "</pre>\n<p>envelope: beat-2s-window probes-max-16 probe-tmo-5s page-12h-720pt ledger-rot-3MiB bench-conc-64-total-100k-loopback-only burst-600x200ms. ledger=knowledge/status/netobs_metrics.log (METRIC protocol + ts=). src=nx_netobs (imports nx_syscalls only). rule-26: /proc reads only, zero hw writes.</p></body></html>\n" as *u8)
  no_atomic("sites/nishifamily/netobs.html" as *u8, pg, o)
  no_w("NETOBS-PAGE ok bytes=" as *u8)
  no_wn(o)
  no_w(" -> sites/nishifamily/netobs.html\n" as *u8)
  return 0
767}
768
769// ---------------- bench (sovereign loadgen; loopback-only BY CONSTRUCTION) ----------------
770// bench CORE (DRY): fork `conc` workers each doing ~total/conc loopback GETs, aggregate latencies, and
771// compute percentiles into res[0..9] = [p50,p90,p95,p99,min,max,rps_milli,ok,err,elapsed_us]. 1 ok / 0 refused
772// (envelope: conc 1..64, total conc..100000). Shared by the `bench` (one-shot) and `ramp` (Step A) verbs.
773func no_bench_run(conc: i64, total: i64, port: i64, path: *u8, res: *i64) -> i64 {
  if conc < 1 { return 0 }
  if conc > NO_BENCH_CONC_MAX { return 0 }
  if total < conc { return 0 }
  if total > NO_BENCH_TOTAL_MAX { return 0 }
  let pids: *i64 = sys_mmap(NO_BENCH_CONC_MAX * 8 + 8) as *i64
  let wp: *u8 = sys_mmap(128)
  let t0: i64 = sys_clock_now_us()
  var wi: i64 = 0
  while wi < conc {
      var per: i64 = total / conc
      if wi == 0 { per = per + (total % conc) }
      let pid: i64 = sys_fork()
      if pid == 0 {
          var po: i64 = no_cat(wp, 0, "/tmp/nxobs_w" as *u8)
          po = no_catn(wp, po, wi)
          po = no_cat(wp, po, ".lat" as *u8)
          wp[po] = 0 as u8
          let wb: *u8 = sys_mmap(NO_MAGIC_2097152)
          let wscr: *u8 = sys_mmap(NO_MAGIC_131072)
          var wo: i64 = 0
          var okc: i64 = 0
          var erc: i64 = 0
          var r: i64 = 0
          while r < per {
              let lat: i64 = no_req(port, path, wscr)
              if lat >= 0 { wo = no_catn(wb, wo, lat); wb[wo] = 10 as u8; wo = wo + 1; okc = okc + 1 } else { erc = erc + 1 }
              r = r + 1
          }
          wo = no_cat(wb, wo, "DONE ok=" as *u8)
          wo = no_catn(wb, wo, okc)
          wo = no_cat(wb, wo, " err=" as *u8)
          wo = no_catn(wb, wo, erc)
          wb[wo] = 10 as u8
          wo = wo + 1
          let wfd: i64 = sys_openat_wr(wp, 420)
          if wfd >= 0 { no_wfd(wfd, wb, wo); sys_close(wfd) }
          sys_exit(0)
      }
      pids[wi] = pid
      wi = wi + 1
  }
  let st: *i64 = sys_mmap(16) as *i64
  var ri: i64 = 0
  while ri < conc { if pids[ri] > 0 { sys_wait4(pids[ri], st, 0) } ri = ri + 1 }
  let t1: i64 = sys_clock_now_us()
  var el: i64 = t1 - t0
  if el < 1 { el = 1 }
  // aggregate (artifact-truth: judge each worker by its file)
  let arr: *i64 = sys_mmap(total * 8 + 64) as *i64
  var cnt: i64 = 0
  var okall: i64 = 0
  var erall: i64 = 0
  let rb: *u8 = sys_mmap(NO_MAGIC_2097152)
  var ai: i64 = 0
  while ai < conc {
      var po: i64 = no_cat(wp, 0, "/tmp/nxobs_w" as *u8)
      po = no_catn(wp, po, ai)
      po = no_cat(wp, po, ".lat" as *u8)
      wp[po] = 0 as u8
      let rn: i64 = no_read(wp, rb, NO_MAGIC_2097151)
      if rn > 0 {
          var li: i64 = 0
          while li < rn {
              let le: i64 = no_line_end(rb, li, rn)
              if le > li {
                  if (rb[li] as i64) == 68 {
                      let seg: *u8 = ((rb as i64) + li) as *u8
                      let sn: i64 = le - li
                      let okv: i64 = no_num_after(seg, sn, "ok=" as *u8)
                      let erv: i64 = no_num_after(seg, sn, "err=" as *u8)
                      if okv >= 0 { okall = okall + okv }
                      if erv >= 0 { erall = erall + erv }
                  } else {
                      let v: i64 = no_span_atoi(rb, li, le)
                      if cnt < total { arr[cnt] = v; cnt = cnt + 1 }
                  }
              }
              li = le + 1
          }
      }
      ai = ai + 1
  }
  no_sort(arr, cnt)
  var mx: i64 = 0 - 1
  var mn2: i64 = 0 - 1
  if cnt > 0 { let ci: i64 = cnt - 1; mx = arr[ci]; mn2 = arr[0] }
  var rpsm: i64 = 0
  if el > 0 { rpsm = (okall * NO_MAGIC_1000000000) / el }
  res[0] = no_pct(arr, cnt, 50)
  res[1] = no_pct(arr, cnt, 90)
  res[2] = no_pct(arr, cnt, 95)
  res[3] = no_pct(arr, cnt, 99)
  res[4] = mn2
  res[5] = mx
  res[6] = rpsm
  res[7] = okall
  res[8] = erall
  res[9] = el
  return 1
873}
874
875// ---------------- ramp (operator Step A: progressive concurrency ramp to saturation) ----------------
876// Ramps concurrency through the declared ladder {1,2,4,8,16,32,64}, each level a bounded bench via the
877// shared core; finds the SATURATION KNEE = the highest conc whose p99 <= target_us AND err==0 (max SUSTAINED
878// load) + the first conc that breaches. Emits the full curve + knee series + log. Loopback-only, envelope-
879// declared (scale law). This IS the operator's Step A ("progressively ramp up until fail/target latencies").
880func no_ramp(port: i64, path: *u8, target_us: i64) -> i64 {
  var tgt: i64 = target_us
  if tgt < 1 { tgt = NO_MAGIC_50000 }
  let levels: *i64 = sys_mmap(64) as *i64
  levels[0] = 1
  levels[1] = 2
  levels[2] = 4
  levels[3] = 8
  levels[4] = 16
  levels[5] = 32
  levels[6] = 64
  let nlev: i64 = 7
  let res: *i64 = sys_mmap(128) as *i64
  let now: i64 = sys_now_realtime_sec()
  let out: *u8 = sys_mmap(NO_MAGIC_65536)
  var o: i64 = no_cat(out, 0, "{\"verb\":\"ramp\",\"target\":\"127.0.0.1:" as *u8)
  o = no_catn(out, o, port)
  o = no_cat(out, o, " " as *u8)
  o = no_cat(out, o, path)
  o = no_cat(out, o, "\",\"ts\":" as *u8)
  o = no_catn(out, o, now)
  o = no_cat(out, o, ",\"target_p99_us\":" as *u8)
  o = no_catn(out, o, tgt)
  o = no_cat(out, o, ",\"curve\":[" as *u8)
  var maxsust: i64 = 0
  var kneeconc: i64 = 0
  var peakrps: i64 = 0
  var emitted: i64 = 0
  var i: i64 = 0
  while i < nlev {
      let c: i64 = levels[i]
      var tot: i64 = c * 200
      if tot > NO_MAGIC_2000 { tot = NO_MAGIC_2000 }
      let ok: i64 = no_bench_run(c, tot, port, path, res)
      if ok == 1 {
          let p99: i64 = res[3]
          let rpsm: i64 = res[6]
          let err: i64 = res[8]
          if emitted > 0 { o = no_cat(out, o, "," as *u8) }
          o = no_cat(out, o, "{\"conc\":" as *u8)
          o = no_catn(out, o, c)
          o = no_cat(out, o, ",\"p50_us\":" as *u8)
          o = no_catn(out, o, res[0])
          o = no_cat(out, o, ",\"p99_us\":" as *u8)
          o = no_catn(out, o, p99)
          o = no_cat(out, o, ",\"rps_milli\":" as *u8)
          o = no_catn(out, o, rpsm)
          o = no_cat(out, o, ",\"err\":" as *u8)
          o = no_catn(out, o, err)
          o = no_cat(out, o, "}" as *u8)
          emitted = emitted + 1
          if rpsm > peakrps { peakrps = rpsm }
          var sustained: i64 = 0
          if p99 <= tgt { if err == 0 { sustained = 1 } }
          if sustained == 1 { maxsust = c } else { if kneeconc == 0 { kneeconc = c } }
      }
      i = i + 1
  }
  o = no_cat(out, o, "],\"max_sustained_conc\":" as *u8)
  o = no_catn(out, o, maxsust)
  o = no_cat(out, o, ",\"knee_conc\":" as *u8)
  o = no_catn(out, o, kneeconc)
  o = no_cat(out, o, ",\"peak_rps_milli\":" as *u8)
  o = no_catn(out, o, peakrps)
  o = no_cat(out, o, ",\"envelope\":{\"ladder\":\"1,2,4,8,16,32,64\",\"per_level_total\":\"min(conc*200,2000)\",\"target\":\"loopback-only\",\"knee\":\"highest conc with p99<=target AND err=0\"}}\n" as *u8)
  sys_write(1, out, o)
  no_atomic("knowledge/status/netobs_ramp.log" as *u8, out, o)
  let lb: *u8 = sys_mmap(NO_MAGIC_4096)
  let tu: i64 = sys_now_us()
  var lo: i64 = no_mrow(lb, 0, "netobs_ramp_max_sustained_conc" as *u8, tu, now, maxsust, "conns" as *u8)
  lo = no_mrow(lb, lo, "netobs_ramp_knee_conc" as *u8, tu, now, kneeconc, "conns" as *u8)
  lo = no_mrow(lb, lo, "netobs_ramp_peak_rps_milli" as *u8, tu, now, peakrps, "millirps" as *u8)
  no_append("knowledge/status/netobs_metrics.log" as *u8, lb, lo)
  no_w("NETOBS-RAMP max_sustained_conc=" as *u8)
  no_wn(maxsust)
  no_w(" knee_conc=" as *u8)
  no_wn(kneeconc)
  no_w(" peak_rps_milli=" as *u8)
  no_wn(peakrps)
  no_w(" target_p99_us=" as *u8)
  no_wn(tgt)
  no_w(" ladder=1,2,4,8,16,32,64 loopback-only VERDICT=GREEN\n" as *u8)
  return 0
963}
964
965// bench (one-shot loadgen; k6 role): run the shared core once + emit JSON + bench.log + ledger.
966func no_bench(conc: i64, total: i64, port: i64, path: *u8) -> i64 {
  let res: *i64 = sys_mmap(128) as *i64
  if no_bench_run(conc, total, port, path, res) == 0 { no_w("NETOBS-BENCH REFUSED (envelope: conc 1..64, total conc..100000)\n" as *u8); sys_exit(3); return 3 }
  let p50: i64 = res[0]
  let p90: i64 = res[1]
  let p95: i64 = res[2]
  let p99: i64 = res[3]
  let mn2: i64 = res[4]
  let mx: i64 = res[5]
  let rpsm: i64 = res[6]
  let okall: i64 = res[7]
  let erall: i64 = res[8]
  let el: i64 = res[9]
  let js: *u8 = sys_mmap(NO_MAGIC_65536)
  var jo: i64 = no_cat(js, 0, "{\"verb\":\"bench\",\"target\":\"127.0.0.1:" as *u8)
  jo = no_catn(js, jo, port)
  jo = no_cat(js, jo, " " as *u8)
  jo = no_cat(js, jo, path)
  jo = no_cat(js, jo, "\",\"ts\":" as *u8)
  jo = no_catn(js, jo, sys_now_realtime_sec())
  jo = no_cat(js, jo, ",\"conc\":" as *u8)
  jo = no_catn(js, jo, conc)
  jo = no_cat(js, jo, ",\"total\":" as *u8)
  jo = no_catn(js, jo, total)
  jo = no_cat(js, jo, ",\"ok\":" as *u8)
  jo = no_catn(js, jo, okall)
  jo = no_cat(js, jo, ",\"err\":" as *u8)
  jo = no_catn(js, jo, erall)
  jo = no_cat(js, jo, ",\"elapsed_us\":" as *u8)
  jo = no_catn(js, jo, el)
  jo = no_cat(js, jo, ",\"p50_us\":" as *u8)
  jo = no_catn(js, jo, p50)
  jo = no_cat(js, jo, ",\"p90_us\":" as *u8)
  jo = no_catn(js, jo, p90)
  jo = no_cat(js, jo, ",\"p95_us\":" as *u8)
  jo = no_catn(js, jo, p95)
  jo = no_cat(js, jo, ",\"p99_us\":" as *u8)
  jo = no_catn(js, jo, p99)
  jo = no_cat(js, jo, ",\"min_us\":" as *u8)
  jo = no_catn(js, jo, mn2)
  jo = no_cat(js, jo, ",\"max_us\":" as *u8)
  jo = no_catn(js, jo, mx)
  jo = no_cat(js, jo, ",\"rps_milli\":" as *u8)
  jo = no_catn(js, jo, rpsm)
  jo = no_cat(js, jo, ",\"pct_method\":\"nearest-rank-floor\",\"envelope\":{\"conc_max\":64,\"total_max\":100000,\"target\":\"loopback-only\",\"conn\":\"close-per-request\",\"drain_cap\":262144,\"timeout_s\":5}}\n" as *u8)
  sys_write(1, js, jo)
  no_atomic("knowledge/status/netobs_bench.log" as *u8, js, jo)
  let lb: *u8 = sys_mmap(NO_MAGIC_4096)
  let tu: i64 = sys_now_us()
  let nw: i64 = sys_now_realtime_sec()
  var lo: i64 = no_mrow(lb, 0, "netobs_bench_p50_us" as *u8, tu, nw, p50, "us" as *u8)
  lo = no_mrow(lb, lo, "netobs_bench_p99_us" as *u8, tu, nw, p99, "us" as *u8)
  lo = no_mrow(lb, lo, "netobs_bench_rps_milli" as *u8, tu, nw, rpsm, "millirps" as *u8)
  lo = no_mrow(lb, lo, "netobs_bench_err" as *u8, tu, nw, erall, "reqs" as *u8)
  no_append("knowledge/status/netobs_metrics.log" as *u8, lb, lo)
  return 0
1022}
1023
1024// ---------------- verdict (analyzer; thresholds are DATA, rule 11) ----------------
1025func no_verdict() -> i64 {
  let man: *u8 = sys_mmap(NO_MAGIC_65536)
  let mn: i64 = no_load_cfg("knowledge/store/netobs-thresholds-" as *u8, "knowledge/registry/netobs_thresholds.tsv" as *u8, man, NO_MAGIC_65535)
  if mn <= 0 { no_w("NETOBS-VERDICT FAIL thresholds manifest unreadable\n" as *u8); sys_exit(2); return 2 }
  let buf: *u8 = sys_mmap(NO_CAP)
  let n: i64 = no_read("knowledge/status/netobs_metrics.log" as *u8, buf, NO_CAP)
  let now: i64 = sys_now_realtime_sec()
  let gl: *u8 = sys_mmap(NO_MAGIC_65536)
  var go2: i64 = 0
  let c0: *i64 = sys_mmap(16) as *i64
  let c1: *i64 = sys_mmap(16) as *i64
  let c2: *i64 = sys_mmap(16) as *i64
  let c3: *i64 = sys_mmap(16) as *i64
  let mname: *u8 = sys_mmap(128)
  let mbound: *u8 = sys_mmap(32)
  let lv: *i64 = sys_mmap(32) as *i64
  var checks: i64 = 0
  var breaches: i64 = 0
  var mi: i64 = 0
  while mi < mn {
      let le: i64 = no_line_end(man, mi, mn)
      if le > mi { if (man[mi] as i64) != 35 {
          if no_col(man, mi, le, 0, c0) == 1 { if no_col(man, mi, le, 1, c1) == 1 { if no_col(man, mi, le, 2, c2) == 1 {
              no_cstr(man, c0[0], c0[1], mname)
              no_cstr(man, c1[0], c1[1], mbound)
              let lim: i64 = no_span_atoi(man, c2[0], c2[1])
              checks = checks + 1
              var have: i64 = 0
              if n > 0 { have = no_last(buf, n, mname, lv) }
              var ok: i64 = 0
              var val: i64 = 0 - 1
              if have == 1 {
                  val = lv[1]
                  if mbound[0] == (109 as u8) { if mbound[1] == (97 as u8) { if val <= lim { ok = 1 } } else { if val >= lim { ok = 1 } } }
              }
              if ok == 0 { breaches = breaches + 1 }
              go2 = no_cat(gl, go2, "CHECK " as *u8)
              go2 = no_cat(gl, go2, mname)
              go2 = no_cat(gl, go2, " value=" as *u8)
              go2 = no_catn(gl, go2, val)
              go2 = no_cat(gl, go2, " bound=" as *u8)
              go2 = no_cat(gl, go2, mbound)
              go2 = no_cat(gl, go2, " limit=" as *u8)
              go2 = no_catn(gl, go2, lim)
              go2 = no_cat(gl, go2, " ok=" as *u8)
              go2 = no_catn(gl, go2, ok)
              if ok == 0 { if no_col(man, mi, le, 3, c3) == 1 { go2 = no_cat(gl, go2, " why=" as *u8); var wi2: i64 = c3[0]; while wi2 < c3[1] { gl[go2] = man[wi2]; go2 = go2 + 1; wi2 = wi2 + 1 } } }
              go2 = no_cat(gl, go2, "\n" as *u8)
          } } }
      } }
      mi = le + 1
  }
  go2 = no_cat(gl, go2, "VERDICT=" as *u8)
  if breaches == 0 { go2 = no_cat(gl, go2, "GREEN" as *u8) } else { go2 = no_cat(gl, go2, "RED" as *u8) }
  go2 = no_cat(gl, go2, " ts=" as *u8)
  go2 = no_catn(gl, go2, now)
  go2 = no_cat(gl, go2, " checks=" as *u8)
  go2 = no_catn(gl, go2, checks)
  go2 = no_cat(gl, go2, " breaches=" as *u8)
  go2 = no_catn(gl, go2, breaches)
  go2 = no_cat(gl, go2, " src=nx_netobs envelope=latest-sample,data-driven-thresholds\n" as *u8)
  no_atomic("knowledge/status/netobs_gate.log" as *u8, gl, go2)
  sys_write(1, gl, go2)
  return 0
1089}
1090
1091// ---------------- selftest ----------------
1092func no_selftest() -> i64 {
  var pass: i64 = 0
  // T1 cpu parse sanity over a 300ms window
  let a2: *i64 = sys_mmap(32) as *i64
  let b2: *i64 = sys_mmap(32) as *i64
  let r1: i64 = no_cpu(a2)
  sys_sleep_ms(300)
  let r2: i64 = no_cpu(b2)
  var t1ok: i64 = 0
  if r1 == 1 { if r2 == 1 { let dtot: i64 = b2[1] - a2[1]; let dbusy: i64 = b2[0] - a2[0]; if dtot > 0 { let pm: i64 = (dbusy * 1000) / dtot; if pm >= 0 { if pm <= 1000 { t1ok = 1 } } } } }
  if t1ok == 1 { pass = pass + 1; no_w("  T1 cpu-parse PASS\n" as *u8) } else { no_w("  T1 cpu-parse FAIL\n" as *u8) }
  // T2 percentile KAT: permutation of 1..100 -> p50==50 p99==99 min==1
  let ka: *i64 = sys_mmap(NO_MAGIC_1024) as *i64
  var ki: i64 = 0
  while ki < 100 { ka[ki] = ((ki * 37) % 100) + 1; ki = ki + 1 }
  no_sort(ka, 100)
  var t2ok: i64 = 0
  let kp50: i64 = no_pct(ka, 100, 50)
  let kp99: i64 = no_pct(ka, 100, 99)
  if kp50 == 50 { if kp99 == 99 { if ka[0] == 1 { if ka[99] == 100 { t2ok = 1 } } } }
  if t2ok == 1 { pass = pass + 1; no_w("  T2 percentile-KAT PASS\n" as *u8) } else { no_w("  T2 percentile-KAT FAIL\n" as *u8) }
  // T3 ledger roundtrip on a scratch path
  let lb: *u8 = sys_mmap(NO_MAGIC_1024)
  var lo: i64 = no_mrow(lb, 0, "netobs_st_probe" as *u8, NO_MAGIC_7777, NO_MAGIC_1111, 7, "kat" as *u8)
  lo = no_mrow(lb, lo, "netobs_st_probe" as *u8, NO_MAGIC_8888, NO_MAGIC_2222, 9, "kat" as *u8)
  let sp: *u8 = "/tmp/nxobs_st.log" as *u8
  let cf: i64 = sys_openat_wr(sp, 420)
  if cf >= 0 { no_wfd(cf, lb, lo); sys_close(cf) }
  let rb2: *u8 = sys_mmap(NO_MAGIC_4096)
  let rn2: i64 = no_read(sp, rb2, NO_MAGIC_4095)
  let lv: *i64 = sys_mmap(32) as *i64
  var t3ok: i64 = 0
  if rn2 > 0 { if no_last(rb2, rn2, "netobs_st_probe" as *u8, lv) == 1 { if lv[0] == NO_MAGIC_2222 { if lv[1] == 9 { t3ok = 1 } } } }
  if t3ok == 1 { pass = pass + 1; no_w("  T3 ledger-roundtrip PASS\n" as *u8) } else { no_w("  T3 ledger-roundtrip FAIL\n" as *u8) }
  // T4 loopback probe against the edge (8443 = sites.elf, always supervised UP)
  let p3: *i64 = sys_mmap(32) as *i64
  let tscr: *u8 = sys_mmap(NO_MAGIC_131072)
  let t4: i64 = no_probe(NO_MAGIC_8443, "/" as *u8, 0, p3, tscr)
  if t4 == 1 { pass = pass + 1; no_w("  T4 probe-connect PASS conn_us=" as *u8); no_wn(p3[0]); no_w("\n" as *u8) } else { no_w("  T4 probe-connect FAIL\n" as *u8) }
  no_w("NETOBS-SELFTEST " as *u8)
  no_wn(pass)
  no_w("/4 VERDICT=" as *u8)
  if pass == 4 { no_w("GREEN\n" as *u8); sys_exit(0); return 0 }
  no_w("RED\n" as *u8)
  sys_exit(1)
  return 1
1138}
1139
1140// one-time no-tsv migration (operator law): read the .tsv config -> sts_seed into the sovereign seg-store
1141// plane -> verify by read-back. After this the .tsv is retired; nx_netobs reads the plane via sts_load.
1142func no_seedcfg() -> i64 {
  let buf: *u8 = sys_mmap(NO_MAGIC_65536)
  let pn: i64 = no_read("knowledge/registry/netobs_probes.tsv" as *u8, buf, NO_MAGIC_65535)
  var pr: i64 = 0 - 1
  if pn > 0 { pr = sts_seed("knowledge/store/netobs-probes-" as *u8, buf, pn) }
  let buf2: *u8 = sys_mmap(NO_MAGIC_65536)
  let tn: i64 = no_read("knowledge/registry/netobs_thresholds.tsv" as *u8, buf2, NO_MAGIC_65535)
  var tr: i64 = 0 - 1
  if tn > 0 { tr = sts_seed("knowledge/store/netobs-thresholds-" as *u8, buf2, tn) }
  let vb: *u8 = sys_mmap(NO_MAGIC_65536)
  let vp: i64 = sts_load("knowledge/store/netobs-probes-" as *u8, vb, NO_MAGIC_65535)
  let vb2: *u8 = sys_mmap(NO_MAGIC_65536)
  let vt: i64 = sts_load("knowledge/store/netobs-thresholds-" as *u8, vb2, NO_MAGIC_65535)
  no_w("NETOBS-SEEDCFG probes_rows=" as *u8)
  no_wn(pr)
  no_w(" probes_plane_bytes=" as *u8)
  no_wn(vp)
  no_w(" thresh_rows=" as *u8)
  no_wn(tr)
  no_w(" thresh_plane_bytes=" as *u8)
  no_wn(vt)
  if pr >= 0 { if tr >= 0 { if vp > 0 { if vt > 0 { no_w(" VERDICT=GREEN\n" as *u8); sys_exit(0); return 0 } } } }
  no_w(" VERDICT=RED\n" as *u8)
  sys_exit(1)
  return 1
1167}
1168
1169func main(argc: i64, argv: *i64) -> i64 {
  if argc < 2 { no_beat(1); return 0 }
  let v: *u8 = argv[1] as *u8
  if no_eq(v, "beat" as *u8) == 1 { no_beat(1); return 0 }
  if no_eq(v, "sample" as *u8) == 1 { no_beat(0); return 0 }
  if no_eq(v, "page" as *u8) == 1 { no_page(); return 0 }
  if no_eq(v, "verdict" as *u8) == 1 { no_verdict(); return 0 }
  if no_eq(v, "selftest" as *u8) == 1 { no_selftest(); return 0 }
  if no_eq(v, "seedcfg" as *u8) == 1 { no_seedcfg(); return 0 }
  if no_eq(v, "ramp" as *u8) == 1 {
      var rp: i64 = NO_MAGIC_18090
      if argc >= 3 { rp = no_atoi(argv[2] as *u8) }
      var rpath: *u8 = "/\x00" as *u8
      if argc >= 4 { rpath = argv[3] as *u8 }
      var rtgt: i64 = NO_MAGIC_50000
      if argc >= 5 { rtgt = no_atoi(argv[4] as *u8) }
      no_ramp(rp, rpath, rtgt)
      return 0
  }
  if no_eq(v, "burst" as *u8) == 1 {
      if argc < 4 { no_w("usage: nx_netobs burst <n<=600> <ms>=200>\n" as *u8); sys_exit(2); return 2 }
      let bn: i64 = no_atoi(argv[2] as *u8)
      let bm: i64 = no_atoi(argv[3] as *u8)
      no_burst(bn, bm)
      return 0
  }
  if no_eq(v, "bench" as *u8) == 1 {
      if argc < 6 { no_w("usage: nx_netobs bench <conc<=64> <total<=100000> <port> <path> (loopback-only)\n" as *u8); sys_exit(2); return 2 }
      let bc: i64 = no_atoi(argv[2] as *u8)
      let bt: i64 = no_atoi(argv[3] as *u8)
      let bp: i64 = no_atoi(argv[4] as *u8)
      no_bench(bc, bt, bp, argv[5] as *u8)
      return 0
  }
  no_w("usage: nx_netobs [beat|sample|page|verdict|selftest|burst n ms|bench conc total port path]\n" as *u8)
  sys_exit(2)
  return 2
1206}
nx_netobs.nx source

`nx_netobs.nx` source
