nx_workstream_popqueue.nx source

1// nx_workstream_popqueue.nx -- WMS rung REG-POP: FULL REGISTRY POPULATION (close the completeness gap). 2// 3// The "never lose an empire again" payoff. WMS-R1 seeded 15 sample streams; the assignment_queue holds 4// ~1251 work-unit ids that the R4 audit reports as UNREGISTERED (prose/queue-only, never in the SSOT). 5// This organ ENUMERATES every queue id and REGISTERS it into the WMS-R1 store, so R4's unregistered 6// count drops from ~1248 toward 0. It does NOT move `untracked` (that counts *.nx organs with no 7// code_link -- queue ids are work-units, not code paths; that is a SEPARATE rung, flagged OPEN here). 8// 9// SOVEREIGN COMMIT DISCIPLINE: ws_put commits ONE new segment per call -> 1248 puts would blow the 10// seg-store 256-segment cap. So this builds ONE writer buffer (ss_begin) holding the merged ws:ids 11// index PLUS every changed ws:<id> record, then ONE ss_commit = ONE new segment. Re-runs that change 12// nothing add ZERO records to the buffer -> ZERO new segments (idempotent; the 256 cap is never 13// threatened). The idempotency engine is a byte-compare against the store's current value (mirrors the 14// seed organ's wss_streq_store skip). 15// 16// EMPIRE DERIVATION (data-driven, rule 11): queue ids carry no empire field, so empire is derived from 17// the id's prefix family via a TABLE (parallel const arrays), and every derived empire is a real member 18// of ws:empires so the R1 internal audit (ws_audit_complete) stays clean. Unknown prefix -> E-MISC. 19// 20// State = FLOATING (queue-only, no measured detail -- honest per the seed doctrine "ABSENT honestly 21// rather than invented"; FLOATING is in the store's documented label set). memory_link/code_link = "-" 22// (intentionally none -> NOT an orphan, per wa_*_link_resolves "-" handling). 23// 24// MANDATORY write discipline: the evidence log line is built into ONE buffer and emitted via the 25// locked atomic fa_appendz (following cn_emit_green/cn_emit_red), with a UNIQUE scratch path 26// (epoch+microsecond) so the organ is safe to run concurrently with itself. The registry writes 27// themselves go through the atomic ss_commit (tmp+fsync+rename commit point). 28// 29// Sovereign: imports only nx_workstream_store + nx_seg_store + nx_framed_append + nx_syscalls (no gcc). 30// Additive-only: does NOT modify any existing organ. license_tier: ORIGINAL 31import "nx_workstream_store.nx" 32import "nx_seg_store.nx" 33import "nx_framed_append.nx" 34import "nx_syscalls.nx" 35 36const POP_LOG: *u8 = "knowledge/status/workstream_popqueue.log" 37const POP_QUEUE: *u8 = "knowledge/registry/assignment_queue.tsv" 38const POP_REC_CAP: i64 = 512 // bounded log record size (no magic number) 39const POP_ID_CAP: i64 = 4000 // hard bound on queue ids collected (JPL rule 2; ~1251 today) 40 41func pq_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n } 42 43// 1 if NUL-term prefix `pre` is a prefix of NUL-term `id`. 44func pq_has_prefix(id: *u8, pre: *u8) -> i64 { 45 var i: i64 = 0 46 while pre[i] != (0 as u8) { 47 if id[i] != pre[i] { return 0 } 48 i = i + 1 49 } 50 return 1 51} 52 53// copy NUL-term src into dst (incl terminator); returns length. 54func pq_cpy(dst: *u8, src: *u8) -> i64 { 55 var i: i64 = 0 56 while src[i] != (0 as u8) { dst[i] = src[i]; i = i + 1 } 57 dst[i] = 0 as u8 58 return i 59} 60 61// ---- data-driven prefix -> empire MAP (rule 11: a table, not buried literals). Longest-meaningful 62// prefix wins by listing more-specific prefixes first. Every target empire is a member of the live 63// ws:empires list (E-CORE/E-ELDER/E-NGE/E-PLATFORM/E-DEPLOY/E-NXC2/E-MISC/E-NISHIOS/E-RESEARCH/ 64// E-DOCTRINE/E-PM/E-WEB). Unknown -> E-MISC (the documented catch-all). ---- 65func pq_empire_for(id: *u8, out: *u8) -> i64 { 66 // queue id families observed in assignment_queue.tsv (col0): 67 // X-B0/X-Q/X-S*/X-* control+backlog rungs -> E-PM (program management) 68 // GEO-* -> E-CORE (sovereign geo stack lives in nishi-core) 69 // WMS-* -> E-CORE (workstream durability substrate) 70 // NBC/NB-/NISH-/H2-/HTTP -> E-WEB (browser/transport/serve) 71 // VAULT-/INFRA-/DEP-/DEPLOY -> E-DEPLOY (infra/deploy/secrets) 72 // GPU-/DXG -> E-PLATFORM (hardware platform reach) 73 // FNET-/SOV-/TRAIN- -> E-RESEARCH (sub-quadratic AI research) 74 // ME2-/OEUV-/MANHEIM-/NAS- -> E-RESEARCH (researcher/analyst arc) 75 // PETS-/NGE- -> E-NGE (games empire) 76 // DOCKER-/PE- -> E-CORE (docker-replacement native stack) 77 if pq_has_prefix(id, "GEO-" as *u8) == 1 { pq_cpy(out, "E-CORE" as *u8); return 0 } 78 if pq_has_prefix(id, "WMS-" as *u8) == 1 { pq_cpy(out, "E-CORE" as *u8); return 0 } 79 if pq_has_prefix(id, "DOCKER-" as *u8) == 1 { pq_cpy(out, "E-CORE" as *u8); return 0 } 80 if pq_has_prefix(id, "PE-" as *u8) == 1 { pq_cpy(out, "E-CORE" as *u8); return 0 } 81 if pq_has_prefix(id, "NBC" as *u8) == 1 { pq_cpy(out, "E-WEB" as *u8); return 0 } 82 if pq_has_prefix(id, "NB-" as *u8) == 1 { pq_cpy(out, "E-WEB" as *u8); return 0 } 83 if pq_has_prefix(id, "NISH-" as *u8) == 1 { pq_cpy(out, "E-WEB" as *u8); return 0 } 84 if pq_has_prefix(id, "H2-" as *u8) == 1 { pq_cpy(out, "E-WEB" as *u8); return 0 } 85 if pq_has_prefix(id, "HTTP" as *u8) == 1 { pq_cpy(out, "E-WEB" as *u8); return 0 } 86 if pq_has_prefix(id, "VAULT-" as *u8) == 1 { pq_cpy(out, "E-DEPLOY" as *u8); return 0 } 87 if pq_has_prefix(id, "INFRA-" as *u8) == 1 { pq_cpy(out, "E-DEPLOY" as *u8); return 0 } 88 if pq_has_prefix(id, "DEP-" as *u8) == 1 { pq_cpy(out, "E-DEPLOY" as *u8); return 0 } 89 if pq_has_prefix(id, "DEPLOY" as *u8) == 1 { pq_cpy(out, "E-DEPLOY" as *u8); return 0 } 90 if pq_has_prefix(id, "GPU-" as *u8) == 1 { pq_cpy(out, "E-PLATFORM" as *u8); return 0 } 91 if pq_has_prefix(id, "DXG" as *u8) == 1 { pq_cpy(out, "E-PLATFORM" as *u8); return 0 } 92 if pq_has_prefix(id, "FNET-" as *u8) == 1 { pq_cpy(out, "E-RESEARCH" as *u8); return 0 } 93 if pq_has_prefix(id, "SOV-" as *u8) == 1 { pq_cpy(out, "E-RESEARCH" as *u8); return 0 } 94 if pq_has_prefix(id, "TRAIN-" as *u8) == 1 { pq_cpy(out, "E-RESEARCH" as *u8); return 0 } 95 if pq_has_prefix(id, "ME2-" as *u8) == 1 { pq_cpy(out, "E-RESEARCH" as *u8); return 0 } 96 if pq_has_prefix(id, "OEUV" as *u8) == 1 { pq_cpy(out, "E-RESEARCH" as *u8); return 0 } 97 if pq_has_prefix(id, "MANHEIM" as *u8) == 1 { pq_cpy(out, "E-RESEARCH" as *u8); return 0 } 98 if pq_has_prefix(id, "NAS-" as *u8) == 1 { pq_cpy(out, "E-RESEARCH" as *u8); return 0 } 99 if pq_has_prefix(id, "PETS-" as *u8) == 1 { pq_cpy(out, "E-NGE" as *u8); return 0 } 100 if pq_has_prefix(id, "NGE-" as *u8) == 1 { pq_cpy(out, "E-NGE" as *u8); return 0 } 101 if pq_has_prefix(id, "X-" as *u8) == 1 { pq_cpy(out, "E-PM" as *u8); return 0 } 102 pq_cpy(out, "E-MISC" as *u8) 103 return 0 104} 105 106// build the 7-field TAB record: id \t empire \t FLOATING \t 0 \t - \t - \t - (NUL-term). Returns length. 107func pq_build_record(id: *u8, empire: *u8, out: *u8) -> i64 { 108 var o: i64 = 0 109 o = fa_cat(out, o, id) 110 out[o] = 9 as u8; o = o + 1 111 o = fa_cat(out, o, empire) 112 o = fa_cat(out, o, "\tFLOATING\t0\t-\t-\t-\x00" as *u8) 113 return o 114} 115 116// build the "ws:" + id key (NUL-term). Returns length. 117func pq_build_key(id: *u8, out: *u8) -> i64 { 118 out[0] = 119 as u8 // 'w' 119 out[1] = 115 as u8 // 's' 120 out[2] = 58 as u8 // ':' 121 var t: i64 = 0 122 while id[t] != (0 as u8) { out[3 + t] = id[t]; t = t + 1 } 123 out[3 + t] = 0 as u8 124 return 3 + t 125} 126 127// 1 if the store value under `prefix` for `key` byte-equals NUL-term `val`. Drives the idempotent skip. 128// One-shot path (used by the seed-style verify); the HOT populate loop uses pq_streq_h below to avoid 129// re-reading every segment file 1251 times. 130func pq_streq_store(prefix: *u8, key: *u8, val: *u8) -> i64 { 131 let pq: *i64 = sys_mmap(16) as *i64 132 let lq: *i64 = sys_mmap(16) as *i64 133 if ss_get(prefix, key, pq, lq) != 1 { return 0 } 134 let b: *u8 = pq[0] as *u8 135 let n: i64 = lq[0] 136 let vl: i64 = pq_slen(val) 137 if n != vl { return 0 } 138 var i: i64 = 0 139 while i < n { if b[i] != val[i] { return 0 } i = i + 1 } 140 return 1 141} 142 143// HANDLE-BASED idempotency compare: 1 iff the OPEN-store snapshot `h` holds `key` byte-equal to `val`. 144// ss_open reads the manifest + every segment's .docs/.idx ONCE; ss_hget then does an in-memory 145// binary search with ZERO per-call file IO. This is the scalable read path -- the per-key ss_get 146// loop mmaps ~100 KB per call (1251 calls => >100 MB of un-freed mmaps), which starves the raw 147// bump-mmap and makes deep lookups read garbage (the 1162-mismatch idempotency bug). One snapshot 148// fixes it: a clean read of the whole registry, used for all 1251 compares. 149func pq_streq_h(h: *i64, key: *u8, val: *u8) -> i64 { 150 let pq: *i64 = sys_mmap(16) as *i64 151 let lq: *i64 = sys_mmap(16) as *i64 152 if ss_hget(h, key, pq, lq) != 1 { return 0 } 153 let b: *u8 = pq[0] as *u8 154 let n: i64 = lq[0] 155 let vl: i64 = pq_slen(val) 156 if n != vl { return 0 } 157 var i: i64 = 0 158 while i < n { if b[i] != val[i] { return 0 } i = i + 1 } 159 return 1 160} 161 162// collect every non-comment col0 id from the queue into ids_out[] (i64 ptrs to mmap'd NUL-term ids). 163// Line model is IDENTICAL to ws_audit_unregistered_p (skip '#'-lines, col0 up to TAB/newline, advance 164// past newline) so the populator registers exactly what the auditor would otherwise flag. Returns count. 165func pq_collect_ids(queue_path: *u8, ids_out: *i64, cap: i64) -> i64 { 166 let lenq: *i64 = sys_mmap(16) as *i64 167 let buf: *u8 = sys_read_file(queue_path, lenq) 168 if buf as i64 == 0 { return 0 - 1 } 169 let n: i64 = lenq[0] 170 var cnt: i64 = 0 171 var i: i64 = 0 172 while i < n { 173 let is_comment: i64 = (buf[i] == (35 as u8)) as i64 // '#' 174 let idbuf: *u8 = sys_mmap(256) 175 var o: i64 = 0 176 if is_comment == 0 { 177 var go: i64 = 1 178 while go == 1 { 179 if i >= n { go = 0 } 180 else { 181 if buf[i] == (9 as u8) { go = 0 } 182 else { if buf[i] == (10 as u8) { go = 0 } else { if o < 255 { idbuf[o] = buf[i]; o = o + 1 } i = i + 1 } } 183 } 184 } 185 idbuf[o] = 0 as u8 186 } 187 // advance i to the byte AFTER the next newline 188 var skip: i64 = 1 189 while skip == 1 { 190 if i >= n { skip = 0 } 191 else { if buf[i] == (10 as u8) { i = i + 1; skip = 0 } else { i = i + 1 } } 192 } 193 if is_comment == 0 { if o > 0 { if cnt < cap { ids_out[cnt] = idbuf as i64; cnt = cnt + 1 } } } 194 } 195 return cnt 196} 197 198// build the merged ws:ids index = existing ws:ids list + any collected id not already a member, 199// into out (NUL-term). Returns out length. Reuses ws_member for whole-token membership. Reads the 200// existing ws:ids from the OPEN-store handle `h` (zero per-call file IO). 201func pq_merge_ids(h: *i64, ids: *i64, nids: i64, out: *u8) -> i64 { 202 var o: i64 = 0 203 // start from the existing ws:ids (if any) 204 let pq: *i64 = sys_mmap(16) as *i64 205 let lq: *i64 = sys_mmap(16) as *i64 206 var cur: *u8 = "" as *u8 207 var curn: i64 = 0 208 if ss_hget(h, "ws:ids" as *u8, pq, lq) == 1 { cur = pq[0] as *u8; curn = lq[0] } 209 var t: i64 = 0 210 while t < curn { out[o] = cur[t]; o = o + 1; t = t + 1 } 211 out[o] = 0 as u8 212 // append each id not already present (whole-token check against the GROWING out buffer) 213 var i: i64 = 0 214 while i < nids { 215 let id: *u8 = ids[i] as *u8 216 if ws_member(out, o, id) == 0 { 217 if o > 0 { out[o] = 9 as u8; o = o + 1 } // TAB separator 218 var j: i64 = 0 219 while id[j] != (0 as u8) { out[o] = id[j]; o = o + 1; j = j + 1 } 220 out[o] = 0 as u8 221 } 222 i = i + 1 223 } 224 return o 225} 226 227// 1 iff NUL-term `a` byte-equals NUL-term `b`. 228func pq_id_eq(a: *u8, b: *u8) -> i64 { 229 var k: i64 = 0 230 while a[k] != (0 as u8) { if a[k] != b[k] { return 0 } k = k + 1 } 231 if b[k] != (0 as u8) { return 0 } 232 return 1 233} 234 235// register ONE id into writer `w` against snapshot `h` (idempotent skip via pq_streq_h). Tallies into 236// out4[1]=registered_new / out4[2]=skipped / out4[3]=errors. Kept as a SMALL function so nx_cc does not 237// hit the register-pressure ceiling that mis-compiles a single oversized populate body (rule 3: split, 238// don't patch). scratch keybuf/recbuf/empbuf are caller-owned (reused across ids). 239func pq_reg_one(h: *i64, w: *i64, id: *u8, keybuf: *u8, recbuf: *u8, empbuf: *u8, out4: *i64) -> i64 { 240 pq_empire_for(id, empbuf) 241 pq_build_key(id, keybuf) 242 let rlen: i64 = pq_build_record(id, empbuf, recbuf) 243 if pq_streq_h(h, keybuf, recbuf) == 1 { 244 out4[2] = out4[2] + 1 // unchanged -> idempotent skip 245 return 0 246 } 247 let rc: i64 = ss_add(w, 1, keybuf, recbuf, rlen) 248 if rc < 0 { out4[3] = out4[3] + 1; return rc } // buffer overflow -> error 249 out4[1] = out4[1] + 1 // staged a new version 250 return 0 251} 252 253// THE CORE: register every collected queue id (except an optional `omit` id, for the gate's neg-control) 254// into the registry under `prefix` in ONE writer buffer / ONE commit. Returns: 255// counters via out4[]: out4[0]=queued out4[1]=registered_new out4[2]=skipped out4[3]=errors 256// function return: 0 on success (commit ok or nothing-to-commit), <0 on a hard commit error. 257// `omit` = NUL-term id to deliberately skip (or 0 for none) -- proves the audit detects a gap. 258// stage the merged ws:ids index into writer `w` iff it changed (idempotent). Small helper. 259func pq_stage_ids(h: *i64, w: *i64, ids: *i64, nids: i64) -> i64 { 260 let mids: *u8 = sys_mmap(1048576) 261 let mlen: i64 = pq_merge_ids(h, ids, nids, mids) 262 if pq_streq_h(h, "ws:ids" as *u8, mids) == 0 { ss_add(w, 1, "ws:ids" as *u8, mids, mlen) } 263 return 0 264} 265 266// stage every per-id ws:<id> record into writer `w` (skipping `omit`). Small helper -- keeps the loop 267// body and its scratch buffers OUT of pq_populate_omit_p so nx_cc never crosses the function-size 268// ceiling that silently mis-compiles a fat body (rule 3: compose small functions, don't patch a big one). 269func pq_stage_records(h: *i64, w: *i64, ids: *i64, nids: i64, omit: *u8, out4: *i64) -> i64 { 270 let keybuf: *u8 = sys_mmap(512) 271 let recbuf: *u8 = sys_mmap(512) 272 let empbuf: *u8 = sys_mmap(64) 273 var has_omit: i64 = 0 274 if omit as i64 != 0 { has_omit = 1 } 275 var i: i64 = 0 276 while i < nids { 277 let id: *u8 = ids[i] as *u8 278 var skip_this: i64 = 0 279 if has_omit == 1 { if pq_id_eq(id, omit) == 1 { skip_this = 1 } } 280 if skip_this == 0 { pq_reg_one(h, w, id, keybuf, recbuf, empbuf, out4) } 281 i = i + 1 282 } 283 return 0 284} 285 286func pq_populate_omit_p(prefix: *u8, queue_path: *u8, omit: *u8, out4: *i64) -> i64 { 287 let ids: *i64 = sys_mmap(8 * POP_ID_CAP) as *i64 288 let nids: i64 = pq_collect_ids(queue_path, ids, POP_ID_CAP) 289 out4[0] = 0; out4[1] = 0; out4[2] = 0; out4[3] = 0 290 if nids < 0 { out4[3] = 1; return 0 - 1 } 291 out4[0] = nids 292 // SNAPSHOT the whole registry ONCE; all idempotency byte-compares read this in-memory handle. 293 let h: *i64 = ss_open(prefix) 294 let w: *i64 = ss_begin() 295 pq_stage_ids(h, w, ids, nids) 296 pq_stage_records(h, w, ids, nids, omit, out4) 297 // ONE commit for the whole batch (one new segment), iff anything was staged. 298 if w[1] > 0 { 299 let segid: i64 = ws_seg_next(prefix) 300 let crc: i64 = ss_commit(prefix, w, segid) 301 if crc != 0 { out4[3] = out4[3] + 1; return crc } 302 } 303 if out4[3] > 0 { return 0 - 2 } 304 return 0 305} 306 307// production convenience: populate the live WS_PREFIX, omit nothing. 308func pq_populate(out4: *i64) -> i64 { 309 return pq_populate_omit_p(WS_PREFIX, POP_QUEUE, 0 as *u8, out4) 310} 311 312// locked-atomic evidence logger (write discipline): ONE assembled buffer -> ONE fa_appendz, with a 313// UNIQUE scratch path so concurrent runs cannot collide. (Here we log directly to POP_LOG via the 314// atomic primitive; the gate uses the same idiom with its own unique scratch path.) 315func pq_log(out4: *i64, verdict_green: i64) -> i64 { 316 let buf: *u8 = sys_mmap(POP_REC_CAP + 16) 317 var o: i64 = 0 318 o = fa_cat(buf, o, "POP authored=organ ts=\x00" as *u8) 319 o = fa_catn(buf, o, sys_now_realtime_sec()) 320 o = fa_cat(buf, o, " queued=\x00" as *u8); o = fa_catn(buf, o, out4[0]) 321 o = fa_cat(buf, o, " registered_new=\x00" as *u8); o = fa_catn(buf, o, out4[1]) 322 o = fa_cat(buf, o, " skipped=\x00" as *u8); o = fa_catn(buf, o, out4[2]) 323 o = fa_cat(buf, o, " errors=\x00" as *u8); o = fa_catn(buf, o, out4[3]) 324 if verdict_green == 1 { o = fa_cat(buf, o, " verdict=GREEN\x00" as *u8) } 325 else { o = fa_cat(buf, o, " verdict=RED\x00" as *u8) } 326 return fa_appendz(POP_LOG, buf, POP_REC_CAP) 327} 328 329// NOTE: this file is a pure LIB (no main) so the gate can import it. The runnable driver lives in 330// nx_workstream_popqueue_run.nx (the harness basename to build+run for a standalone population).