`nx_ws_board.nx` source

↩ module page · 527 lines · 25893 B
1// nx_ws_board.nx -- WMS rung M2: the LIVE STATUS BOARD (derived view).
2//
3// module: nishi-core.wms.ws_board
4// capability: CORE_COMPUTE (a derived, self-validating view over WMS-R1 + WMS-R2)
5//
6// WHAT THIS CLOSES: crash recovery used to mean reading a stale THREADS worksheet
7// and a lossy MEMORY index by hand.  This organ DERIVES the live state of every
8// empire/stream straight from the SSOT registry (WMS-R1, nx_workstream_store) and
9// validates freshness against the transition reflog (WMS-R2, nx_ws_ledger).
10// Crash-recovery becomes "read the board" -- and the board is NEVER silently wrong:
11//   * if the registry is INCOMPLETE (a lost/never-committed segment) it REFUSES to
12//     render (returns -1), it does not paper over the gap.  (ws_audit_complete==0
13//     is a precondition, in the DATA PATH, not just the test.)
14//   * if the reflog overlay is TORN/OLD (replay flags any malformed line) it
15//     REJECTS the reflog-derived view (returns -3) rather than rendering it wrong.
16//
17// IMPEDANCE NOTE (load-bearing, Rule 4 -- every displayed fact is a real field read):
18//   R1 records store `state` as a STRING label (ACTIVE|DONE|BLOCKED|FLOATING|ABSENT).
19//   R2's reflog stores `new=` as an INTEGER (0..4).  The board's PRIMARY source of
20//   truth for displayed state is the R1 string field (ws_field f=2).  The R2 reflog
21//   is used ONLY as an integrity/freshness OVERLAY (replay must yield flagged==0).
22//   We do NOT re-map int->label here -- that int<->label bridge is a future rung; a
23//   fabricated mapping would violate no-false-green.
24//
25// SCHEMA the board surfaces (Rule 4 -- closest VERIFIED facts, no invented columns):
26//   field 1=empire 2=state 3=last_touched 5=code_link (owner/next-rung surrogate).
27//   The registry has no dedicated owner/next_rung columns; code_link(5)+deps(6) are
28//   the nearest measured facts.  Inventing owner/next_rung would be a schema lie.
29//
30// REUSE / lineage: ws_manifest_p / ws_get_p / ws_field / ws_member / ws_audit_complete_p
31// (nx_workstream_store, WMS-R1); ledger_replay (nx_ws_ledger, WMS-R2); fa_appendz /
32// fa_cat / fa_catn (nx_framed_append, WMS-R0b -- the MANDATORY one-buffer-one-locked-
33// write discipline for the file channel).  Additive: no organ modified.  Sovereign:
34// imports only the WMS stack + nx_syscalls (no gcc/openssl/node).  license_tier: ORIGINAL
35import "nx_workstream_store.nx"
36import "nx_itoa_lib.nx"   // shared MSB-first emitter (zero-alloc)
37import "nx_ws_ledger.nx"
38import "nx_framed_append.nx"
39import "nx_heartbeat_str.nx"
40import "nx_syscalls.nx"
41
42const WB_LOG:     *u8 = "knowledge/status/ws_board.log"
43const WB_RECCAP:  i64 = 512        // bounded board-record size
44const WB_MAXEMP:  i64 = 64         // max empires the board rolls up (12 today)
45const WB_MAXRW:   i64 = 64         // reflog state[] capacity (ledger_replay maxws)
46// RETIRED: the old WB_MAXWS=512 stream cap. The board now COUNTS ws:ids and enumerates
47// EVERY registered stream from a single ss_open snapshot -- the registry SSOT sets the
48// size, never a magic number, and never a silent truncation (the bug this rung killed).
49
50// ---- liveness / freshness (Cardinal 11/17: data-driven, never baked) ------------------
51// The board refuses to trust a stale ACTIVE label: liveness is recomputed EVERY render
52// from the HARDWARE clock (sys_now_realtime_sec) vs the stream's last activity = max of
53// registry last_touched and the live heartbeat channel, against a CONFIG window. So
54// "active" means beating NOW (fresh like energy), not a one-time stamp.
55const WB_HB_CHANNEL: *u8 = "knowledge/status/wms_heartbeat.log"   // LIVE-1 beat channel
56const WB_CFG_FRESH_KEY: *u8 = "ws:cfg:fresh_window_sec"           // operator override (DATA)
57const WB_FRESH_WINDOW_DEFAULT_SEC: i64 = 86400  // bootstrap FLOOR only (Cardinal 17 lowest tier)
58const WB_SW:         i64 = 6        // i64 slots per named live-stream row
59const WB_ST_ACTIVE:  i64 = 1
60const WB_ST_BLOCKED: i64 = 2
61
62// derived-view return codes (data-driven; no magic numbers).
63const WB_INCOMPLETE: i64 = 0 - 1   // registry audit failed -> board NOT rendered
64const WB_NOIDS:      i64 = 0 - 2   // ws:ids absent -> nothing to render
65const WB_TORNREFLOG: i64 = 0 - 3   // reflog overlay torn/old -> view REJECTED
66
67// per-empire rollup row, packed into rows_out[] as 6 consecutive i64 slots:
68//   [base+0]=empire-id ptr  +1=total  +2=done  +3=active  +4=blocked  +5=other
69const WB_ROWW: i64 = 6             // i64 slots per empire row
70
71func wb_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
72
73// NUL-terminated byte-equality (state-label compare against the data-driven label set).
74func wb_streq(a: *u8, b: *u8) -> i64 {
  var i: i64 = 0
  while a[i] != (0 as u8) {
      if a[i] != b[i] { return 0 }
      i = i + 1
  }
  if b[i] != (0 as u8) { return 0 }
  return 1
82}
83
84// decimal NUL-terminated string -> i64 (last_touched / config parse; ignores non-digits).
85func wb_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
94}
95
96// COUNT the registered ids straight from the ws:ids snapshot (the data-driven size that
97// replaces the retired 512 cap). Counts TAB-bounded non-empty tokens. ss_open snapshot.
98func wb_count_ids(prefix: *u8) -> i64 {
  let h: *i64 = ss_open(prefix)
  let pq: *i64 = sys_mmap(16) as *i64
  let lq: *i64 = sys_mmap(16) as *i64
  if ss_hget(h, "ws:ids" as *u8, pq, lq) != 1 { return 0 }
  let ids: *u8 = pq[0] as *u8
  let idn: i64 = lq[0]
  var cnt: i64 = 0
  var i: i64 = 0
  while i < idn {
      var atstart: i64 = 0
      if i == 0 { atstart = 1 }
      if i > 0 { if ids[i - 1] == (9 as u8) { atstart = 1 } }
      if atstart == 1 { if ids[i] != (9 as u8) { cnt = cnt + 1 } }
      i = i + 1
  }
  return cnt
115}
116
117// CONFIG read (Cardinal 17 hierarchy): the live freshness window is DATA in the store
118// (ws:cfg:fresh_window_sec). Returns the stored int, or `defval` (the bootstrap floor)
119// if the key is absent -- so the operative number lives in data, not buried in code.
120func wb_cfg_int(prefix: *u8, key: *u8, defval: i64) -> i64 {
  let h: *i64 = ss_open(prefix)
  let pq: *i64 = sys_mmap(16) as *i64
  let lq: *i64 = sys_mmap(16) as *i64
  if ss_hget(h, key, pq, lq) != 1 { return defval }
  let b: *u8 = pq[0] as *u8
  let n: i64 = lq[0]
  if n <= 0 { return defval }
  var v: i64 = 0
  var i: i64 = 0
  while i < n {
      let c: i64 = b[i] as i64
      if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } }
      i = i + 1
  }
  return v
136}
137
138// THE DERIVED VIEW.  Parameterized by prefix so the gate can drive a HERMETIC store.
139// `refpath` (or 0 to skip) is the WMS-R2 reflog used as a freshness/integrity overlay.
140//   rows_out[]   : caller-mmapped, >= WB_MAXEMP*WB_ROWW i64 slots (per-empire rollup).
141//   empcount_out : set to the number of empires rolled up.
142//   summary_out[]: [0]=total_streams [1]=total_done [2]=total_active [3]=total_blocked
143//                  [4]=reflog_flagged (0 when refpath==0 or clean).
144// Returns the empire count (>=0) on success, or a negative WB_* refusal code.
145func wb_board_p(prefix: *u8, refpath: *u8,
              rows_out: *i64, cap: i64,
              empcount_out: *i64, summary_out: *i64) -> i64 {
  // ---- (1) TRUST GATE: registry must be COMPLETE before we trust any of it. ----
  let missing: *i64 = sys_mmap(8 * 128) as *i64
  let mc: i64 = ws_audit_complete_p(prefix, missing, 128)
  if mc < 0 { return WB_NOIDS }       // ws:ids absent
  if mc > 0 { return WB_INCOMPLETE }  // a lost/never-committed segment -> refuse
153
  // ---- (2) REFLOG INTEGRITY OVERLAY: a torn/old reflog REJECTS the view. ----
  var reflog_flagged: i64 = 0
  if refpath != (0 as *u8) {
      let rstate: *i64 = sys_mmap(8 * WB_MAXRW) as *i64
      var ri: i64 = 0
      while ri < WB_MAXRW { rstate[ri] = 0 - 1; ri = ri + 1 }
      let routs: *i64 = sys_mmap(32) as *i64
      ledger_replay(refpath, rstate, routs, WB_MAXRW)
      reflog_flagged = routs[2]
      if reflog_flagged > 0 { return WB_TORNREFLOG }
  }
165
  // ---- (3) snapshot the registry ONCE (ss_open) -> O(1) in-memory ss_hgets.
  // SCALE FIX (the AUDIT-SCALE lesson + Cardinal 21): the old per-id ss_get re-read
  // EVERY segment per lookup (~100KB mmap each) and starved the bump-mmap past ~1162
  // ids. Mirror ws_audit_complete_p: open once, ss_hget thereafter. THIS is what makes
  // retiring the 512 enumeration cap safe at full scale. ----
  let h: *i64 = ss_open(prefix)
  let eq: *i64 = sys_mmap(16) as *i64
  let el: *i64 = sys_mmap(16) as *i64
  var emp: *u8 = "" as *u8
  var empn: i64 = 0
  if ss_hget(h, "ws:empires" as *u8, eq, el) == 1 { emp = eq[0] as *u8; empn = el[0] }
177
  // build the empire-id list (NUL-term ptrs) by TAB-walking ws:empires, and zero its row.
  var ecount: i64 = 0
  var ei: i64 = 0
  while ei < empn {
      // start of a field iff ei==0 or previous byte is TAB
      var atstart: i64 = 0
      if ei == 0 { atstart = 1 }
      if ei > 0 { if emp[ei - 1] == (9 as u8) { atstart = 1 } }
      if atstart == 1 {
          let nbuf: *u8 = sys_mmap(64)
          var o: i64 = 0
          var p: i64 = ei
          var go: i64 = 1
          while go == 1 {
              if p >= empn { go = 0 } else {
                  if emp[p] == (9 as u8) { go = 0 } else { nbuf[o] = emp[p]; o = o + 1; p = p + 1 }
              }
          }
          nbuf[o] = 0 as u8
          if o > 0 { if ecount < WB_MAXEMP {
              let base: i64 = ecount * WB_ROWW
              rows_out[base + 0] = nbuf as i64
              rows_out[base + 1] = 0
              rows_out[base + 2] = 0
              rows_out[base + 3] = 0
              rows_out[base + 4] = 0
              rows_out[base + 5] = 0
              ecount = ecount + 1
          } }
      }
      ei = ei + 1
  }
210
  // ---- (4) enumerate EVERY registered id with NO cap: walk the whole ws:ids string
  // straight from the snapshot, ss_hget each record, tally per empire by real fields. ----
  let pq: *i64 = sys_mmap(16) as *i64
  let lq: *i64 = sys_mmap(16) as *i64
  var idstr: *u8 = "" as *u8
  var idn: i64 = 0
  if ss_hget(h, "ws:ids" as *u8, pq, lq) == 1 { idstr = pq[0] as *u8; idn = lq[0] }
  let rq: *i64 = sys_mmap(16) as *i64
  let rl: *i64 = sys_mmap(16) as *i64
  let keybuf: *u8 = sys_mmap(160)
  let empbuf: *u8 = sys_mmap(64)
  let stbuf:  *u8 = sys_mmap(64)
  var t_total: i64 = 0
  var t_done: i64 = 0
  var t_active: i64 = 0
  var t_blocked: i64 = 0
  var ii: i64 = 0
  while ii < idn {
      let idbuf: *u8 = sys_mmap(160)
      var io: i64 = 0
      var igo: i64 = 1
      while igo == 1 {
          if ii >= idn { igo = 0 } else {
              if idstr[ii] == (9 as u8) { ii = ii + 1; igo = 0 } else { idbuf[io] = idstr[ii]; io = io + 1; ii = ii + 1 }
          }
      }
      idbuf[io] = 0 as u8
      if io > 0 {
          keybuf[0] = 119 as u8; keybuf[1] = 115 as u8; keybuf[2] = 58 as u8
          var z: i64 = 0
          while z < io { keybuf[3 + z] = idbuf[z]; z = z + 1 }
          keybuf[3 + io] = 0 as u8
          if ss_hget(h, keybuf, rq, rl) == 1 {
              ws_field(rq[0] as *u8, rl[0], 1, empbuf)   // empire
              ws_field(rq[0] as *u8, rl[0], 2, stbuf)    // state label
              t_total = t_total + 1
              var is_done: i64 = 0
              var is_active: i64 = 0
              var is_blocked: i64 = 0
              if wb_streq(stbuf, "DONE" as *u8) == 1 { is_done = 1; t_done = t_done + 1 }
              if wb_streq(stbuf, "ACTIVE" as *u8) == 1 { is_active = 1; t_active = t_active + 1 }
              if wb_streq(stbuf, "BLOCKED" as *u8) == 1 { is_blocked = 1; t_blocked = t_blocked + 1 }
              // find the empire's row and bump its tallies
              var e: i64 = 0
              while e < ecount {
                  let base: i64 = e * WB_ROWW
                  if wb_streq(rows_out[base + 0] as *u8, empbuf) == 1 {
                      rows_out[base + 1] = rows_out[base + 1] + 1
                      if is_done == 1 { rows_out[base + 2] = rows_out[base + 2] + 1 }
                      if is_active == 1 { rows_out[base + 3] = rows_out[base + 3] + 1 }
                      if is_blocked == 1 { rows_out[base + 4] = rows_out[base + 4] + 1 }
                      if is_done == 0 { if is_active == 0 { if is_blocked == 0 {
                          rows_out[base + 5] = rows_out[base + 5] + 1
                      } } }
                      e = ecount
                  } else { e = e + 1 }
              }
          }
      }
  }
271
  summary_out[0] = t_total
  summary_out[1] = t_done
  summary_out[2] = t_active
  summary_out[3] = t_blocked
  summary_out[4] = reflog_flagged
  *empcount_out = ecount
  return ecount
279}
280
281// convenience: live ws- prefix.
282func wb_board(refpath: *u8, rows_out: *i64, cap: i64,
            empcount_out: *i64, summary_out: *i64) -> i64 {
  return wb_board_p(WS_PREFIX, refpath, rows_out, cap, empcount_out, summary_out)
285}
286
287// MANDATORY WRITE DISCIPLINE: assemble the WHOLE board line into ONE buffer, then a
288// SINGLE atomic locked fa_appendz to the file channel (cn_emit_red/green pattern).
289// One BOARD line per empire.  Returns fa_appendz rc.
290func wb_emit_empire(path: *u8, empire: *u8, total: i64, done: i64, active: i64, blocked: i64) -> i64 {
  let buf: *u8 = sys_mmap(WB_RECCAP + 16)
  var o: i64 = 0
  o = fa_cat(buf, o, "BOARD empire=\x00" as *u8)
  o = fa_cat(buf, o, empire)
  o = fa_cat(buf, o, " streams=\x00" as *u8)
  o = fa_catn(buf, o, total)
  o = fa_cat(buf, o, " done=\x00" as *u8)
  o = fa_catn(buf, o, done)
  o = fa_cat(buf, o, " active=\x00" as *u8)
  o = fa_catn(buf, o, active)
  o = fa_cat(buf, o, " blocked=\x00" as *u8)
  o = fa_catn(buf, o, blocked)
  o = fa_cat(buf, o, " END\x00" as *u8)
  buf[o] = 0 as u8
  return fa_appendz(path, buf, WB_RECCAP)
306}
307
308// final summary line, one buffer one locked write.
309func wb_emit_summary(path: *u8, total: i64, done: i64, active: i64, blocked: i64, flagged: i64) -> i64 {
  let buf: *u8 = sys_mmap(WB_RECCAP + 16)
  var o: i64 = 0
  o = fa_cat(buf, o, "BOARD-SUMMARY total=\x00" as *u8)
  o = fa_catn(buf, o, total)
  o = fa_cat(buf, o, " done=\x00" as *u8)
  o = fa_catn(buf, o, done)
  o = fa_cat(buf, o, " active=\x00" as *u8)
  o = fa_catn(buf, o, active)
  o = fa_cat(buf, o, " blocked=\x00" as *u8)
  o = fa_catn(buf, o, blocked)
  o = fa_cat(buf, o, " reflog_flagged=\x00" as *u8)
  o = fa_catn(buf, o, flagged)
  o = fa_cat(buf, o, " verdict=RENDERED END\x00" as *u8)
  buf[o] = 0 as u8
  return fa_appendz(path, buf, WB_RECCAP)
325}
326
327// LIVE decomposition line: the active/blocked counts split by FRESH liveness, plus the
328// now+window the verdict was computed against (so the log records WHY a count is what it
329// is). One buffer, one locked atomic fa_appendz.
330func wb_emit_live(path: *u8, now: i64, window: i64, lsm: *i64) -> i64 {
  let buf: *u8 = sys_mmap(WB_RECCAP + 16)
  var o: i64 = 0
  o = fa_cat(buf, o, "BOARD-LIVE now=\x00" as *u8);     o = fa_catn(buf, o, now)
  o = fa_cat(buf, o, " window=\x00" as *u8);            o = fa_catn(buf, o, window)
  o = fa_cat(buf, o, " active=\x00" as *u8);            o = fa_catn(buf, o, lsm[0])
  o = fa_cat(buf, o, " a_alive=\x00" as *u8);           o = fa_catn(buf, o, lsm[1])
  o = fa_cat(buf, o, " a_stalled=\x00" as *u8);         o = fa_catn(buf, o, lsm[2])
  o = fa_cat(buf, o, " a_unknown=\x00" as *u8);         o = fa_catn(buf, o, lsm[3])
  o = fa_cat(buf, o, " blocked=\x00" as *u8);           o = fa_catn(buf, o, lsm[4])
  o = fa_cat(buf, o, " b_alive=\x00" as *u8);           o = fa_catn(buf, o, lsm[5])
  o = fa_cat(buf, o, " b_stalled=\x00" as *u8);         o = fa_catn(buf, o, lsm[6])
  o = fa_cat(buf, o, " b_unknown=\x00" as *u8);         o = fa_catn(buf, o, lsm[7])
  o = fa_cat(buf, o, " END\x00" as *u8)
  buf[o] = 0 as u8
  return fa_appendz(path, buf, WB_RECCAP)
346}
347
348// stdout (fd 1) writers -- terminal, uncontended, plain writes are fine here.
349func wb_p(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
350// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
351// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
352// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
353// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
354func wb_pn(v: i64) -> i64 { nxi_out(v); return 0 }
355
356// THE LIVE-STREAM NAMER: enumerate every registered stream (NO cap, ss_open snapshot)
357// and, for each one whose state is ACTIVE or BLOCKED, record its id + a FRESH liveness
358// verdict. Liveness = compare `now` (the caller injects the hardware clock; the gate
359// injects a controlled epoch -- no hidden wall-clock, mirroring hbm_scan's discipline)
360// against the stream's last activity = max(registry last_touched, last heartbeat beat),
361// using `fresh_window` (the caller sources it from config). So a stream LABELED active but
362// silent for longer than the window classifies STALLED, never silently counted alive.
363//   streams_out[] packed WB_SW i64/row: +0 id ptr +1 statecode(1=ACTIVE 2=BLOCKED)
364//     +2 liveness(HBS_ALIVE/STALLED/UNKNOWN) +3 last_activity +4 age +5 empire ptr
365//   live_summary[]: [0]=active_total [1]=a_alive [2]=a_stalled [3]=a_unknown
366//                   [4]=blocked_total [5]=b_alive [6]=b_stalled [7]=b_unknown
367// Returns the number of named (active+blocked) streams.
368func wb_live_streams_p(prefix: *u8, hbpath: *u8, now: i64, fresh_window: i64,
                     streams_out: *i64, scap: i64, live_summary: *i64) -> i64 {
  var z0: i64 = 0
  while z0 < 8 { live_summary[z0] = 0; z0 = z0 + 1 }
  let h: *i64 = ss_open(prefix)
  let pq: *i64 = sys_mmap(16) as *i64
  let lq: *i64 = sys_mmap(16) as *i64
  var idstr: *u8 = "" as *u8
  var idn: i64 = 0
  if ss_hget(h, "ws:ids" as *u8, pq, lq) == 1 { idstr = pq[0] as *u8; idn = lq[0] }
  let rq: *i64 = sys_mmap(16) as *i64
  let rl: *i64 = sys_mmap(16) as *i64
  let keybuf: *u8 = sys_mmap(160)
  let stbuf:  *u8 = sys_mmap(64)
  let ltbuf:  *u8 = sys_mmap(64)
  var named: i64 = 0
  var ii: i64 = 0
  while ii < idn {
      let idbuf: *u8 = sys_mmap(160)
      var io: i64 = 0
      var igo: i64 = 1
      while igo == 1 {
          if ii >= idn { igo = 0 } else {
              if idstr[ii] == (9 as u8) { ii = ii + 1; igo = 0 } else { idbuf[io] = idstr[ii]; io = io + 1; ii = ii + 1 }
          }
      }
      idbuf[io] = 0 as u8
      if io > 0 {
          keybuf[0] = 119 as u8; keybuf[1] = 115 as u8; keybuf[2] = 58 as u8
          var z: i64 = 0
          while z < io { keybuf[3 + z] = idbuf[z]; z = z + 1 }
          keybuf[3 + io] = 0 as u8
          if ss_hget(h, keybuf, rq, rl) == 1 {
              ws_field(rq[0] as *u8, rl[0], 2, stbuf)   // state label
              var sc: i64 = 0
              if wb_streq(stbuf, "ACTIVE" as *u8) == 1 { sc = WB_ST_ACTIVE }
              if wb_streq(stbuf, "BLOCKED" as *u8) == 1 { sc = WB_ST_BLOCKED }
              if sc != 0 {
                  ws_field(rq[0] as *u8, rl[0], 3, ltbuf)   // last_touched
                  let lt: i64 = wb_atoi(ltbuf)
                  let lb: i64 = hb_last_beat_s(hbpath, idbuf)   // live heartbeat (-1 = none)
                  var last_act: i64 = lt
                  if lb > last_act { last_act = lb }
                  var live: i64 = HBS_UNKNOWN
                  var age: i64 = 0 - 1
                  if last_act > 0 {
                      age = now - last_act
                      if age > fresh_window { live = HBS_STALLED } else { live = HBS_ALIVE }
                  }
                  let empbuf: *u8 = sys_mmap(64)
                  ws_field(rq[0] as *u8, rl[0], 1, empbuf)   // empire
                  if named < scap {
                      let base: i64 = named * WB_SW
                      streams_out[base + 0] = idbuf as i64
                      streams_out[base + 1] = sc
                      streams_out[base + 2] = live
                      streams_out[base + 3] = last_act
                      streams_out[base + 4] = age
                      streams_out[base + 5] = empbuf as i64
                  }
                  named = named + 1
                  if sc == WB_ST_ACTIVE {
                      live_summary[0] = live_summary[0] + 1
                      if live == HBS_ALIVE   { live_summary[1] = live_summary[1] + 1 }
                      if live == HBS_STALLED { live_summary[2] = live_summary[2] + 1 }
                      if live == HBS_UNKNOWN { live_summary[3] = live_summary[3] + 1 }
                  }
                  if sc == WB_ST_BLOCKED {
                      live_summary[4] = live_summary[4] + 1
                      if live == HBS_ALIVE   { live_summary[5] = live_summary[5] + 1 }
                      if live == HBS_STALLED { live_summary[6] = live_summary[6] + 1 }
                      if live == HBS_UNKNOWN { live_summary[7] = live_summary[7] + 1 }
                  }
              }
          }
      }
  }
  return named
446}
447
448// main(): render the LIVE board (ws- registry) to stdout + ws_board.log (file channel
449// via fa_appendz).  The live reflog overlay is skipped (refpath=0) because the live
450// transition log path is not guaranteed present in every environment -- the overlay is
451// an OPTIONAL freshness check; the registry is the SSOT.  Exit 0 on render, non-zero if
452// the registry guard refuses (the loud crash-recovery signal).
453func main() -> i64 {
  wb_p("=== WMS-M2 LIVE STATUS BOARD (derived from WMS-R1 registry + WMS-R2 reflog) ===\n" as *u8)
  let rows: *i64 = sys_mmap(8 * WB_MAXEMP * WB_ROWW) as *i64
  let ec: *i64 = sys_mmap(16) as *i64
  let sm: *i64 = sys_mmap(64) as *i64
  let rc: i64 = wb_board(0 as *u8, rows, WB_MAXEMP, ec, sm)
  if rc < 0 {
      wb_p(" BOARD REFUSED rc=\x00" as *u8); wb_pn(rc)
      if rc == WB_INCOMPLETE { wb_p(" reason=registry-incomplete (lost segment) -- NOT rendered\n" as *u8) }
      if rc == WB_NOIDS { wb_p(" reason=no-ws:ids -- nothing registered\n" as *u8) }
      if rc == WB_TORNREFLOG { wb_p(" reason=torn/old-reflog -- view REJECTED\n" as *u8) }
      sys_exit(1); return 1
  }
  let necount: i64 = ec[0]
  var e: i64 = 0
  while e < necount {
      let base: i64 = e * WB_ROWW
      let empire: *u8 = rows[base + 0] as *u8
      let total: i64 = rows[base + 1]
      let done: i64 = rows[base + 2]
      let active: i64 = rows[base + 3]
      let blocked: i64 = rows[base + 4]
      // stdout line
      wb_p("  empire=\x00" as *u8); wb_p(empire)
      wb_p(" streams=\x00" as *u8); wb_pn(total)
      wb_p(" done=\x00" as *u8); wb_pn(done)
      wb_p(" active=\x00" as *u8); wb_pn(active)
      wb_p(" blocked=\x00" as *u8); wb_pn(blocked)
      wb_p("\n\x00" as *u8)
      // file channel: one locked atomic write per empire
      wb_emit_empire(WB_LOG, empire, total, done, active, blocked)
      e = e + 1
  }
  wb_p("  SUMMARY total=\x00" as *u8); wb_pn(sm[0])
  wb_p(" done=\x00" as *u8); wb_pn(sm[1])
  wb_p(" active=\x00" as *u8); wb_pn(sm[2])
  wb_p(" blocked=\x00" as *u8); wb_pn(sm[3])
  wb_p(" verdict=RENDERED\n\x00" as *u8)
  wb_emit_summary(WB_LOG, sm[0], sm[1], sm[2], sm[3], sm[4])
492
  // ===== LIVE STREAMS: name every active/blocked stream + FRESH liveness =====
  // This is the crash-recovery payoff: not just counts, but WHICH streams, and
  // whether each is actually beating NOW. Recomputed from the hardware clock each
  // render (fresh like energy), with the freshness window sourced from config.
  let now: i64 = sys_now_realtime_sec()
  let fw: i64 = wb_cfg_int(WS_PREFIX, WB_CFG_FRESH_KEY, WB_FRESH_WINDOW_DEFAULT_SEC)
  let nids: i64 = wb_count_ids(WS_PREFIX)
  let streams: *i64 = sys_mmap(8 * WB_SW * (nids + 16)) as *i64
  let lsm: *i64 = sys_mmap(8 * 16) as *i64
  let nnamed: i64 = wb_live_streams_p(WS_PREFIX, WB_HB_CHANNEL, now, fw, streams, nids + 16, lsm)
  wb_p("=== LIVE STREAMS (active/blocked; liveness FRESH @ now=\x00" as *u8); wb_pn(now)
  wb_p(" window=\x00" as *u8); wb_pn(fw); wb_p("s) ===\n\x00" as *u8)
  var si: i64 = 0
  while si < nnamed {
      let sb: i64 = si * WB_SW
      wb_p("  id=\x00" as *u8); wb_p(streams[sb + 0] as *u8)
      if streams[sb + 1] == WB_ST_ACTIVE { wb_p(" state=ACTIVE \x00" as *u8) } else { wb_p(" state=BLOCKED\x00" as *u8) }
      let lv: i64 = streams[sb + 2]
      if lv == HBS_ALIVE   { wb_p(" live=ALIVE  \x00" as *u8) }
      if lv == HBS_STALLED { wb_p(" live=STALLED\x00" as *u8) }
      if lv == HBS_UNKNOWN { wb_p(" live=UNKNOWN\x00" as *u8) }
      wb_p(" age=\x00" as *u8); wb_pn(streams[sb + 4])
      wb_p("s empire=\x00" as *u8); wb_p(streams[sb + 5] as *u8)
      wb_p("\n\x00" as *u8)
      si = si + 1
  }
  wb_p("  LIVE-SUMMARY active=\x00" as *u8); wb_pn(lsm[0])
  wb_p(" (alive=\x00" as *u8); wb_pn(lsm[1]); wb_p(" stalled=\x00" as *u8); wb_pn(lsm[2]); wb_p(" unknown=\x00" as *u8); wb_pn(lsm[3]); wb_p(")\x00" as *u8)
  wb_p(" blocked=\x00" as *u8); wb_pn(lsm[4])
  wb_p(" (alive=\x00" as *u8); wb_pn(lsm[5]); wb_p(" stalled=\x00" as *u8); wb_pn(lsm[6]); wb_p(" unknown=\x00" as *u8); wb_pn(lsm[7]); wb_p(")\n\x00" as *u8)
  wb_emit_live(WB_LOG, now, fw, lsm)
524
  sys_exit(0)
  return 0
527}
nx_ws_board.nx source

`nx_ws_board.nx` source
