nx_ws_ingest.nx source

1// nx_ws_ingest.nx -- THE MISSING INTEGRATION KEYSTONE (WMS derive-from-SSOT R1, for real). 2// 3// The operator's "we shouldn't have lots of separate systems" bug, named and fixed: the machine 4// registry (knowledge/store/ws-, read by nx_ws_board) only ever tracked the assignment-queue ids 5// (via nx_workstream_popqueue) and the PM/old WMS streams. The 426 durable project-*.md arcs -- 6// genealogy, login, media, reader, brand, APM, doctor, ... -- were NEVER ingested, so the board 7// could not see them. nx_ws_manifest_emit only rewrites a MARKDOWN list; it has zero ws_put. This 8// organ is the half that was missing: it SCANS every project-*.md and ws_put's each into the 9// registry so the board ENUMERATES every arc -> one source of truth, not three. 10// 11// REUSE / lineage: the dir walk is wme_scan (nx_ws_index_lib) -- names[] + REAL mtimes[]. The 12// one-snapshot(ss_open)/one-writer(ss_begin)/one-commit(ss_commit) batch discipline + idempotent 13// byte-compare skip is the proven nx_workstream_popqueue path (256-segment-cap safe at >1000 14// records; re-runs add ZERO segments). last_touched = the file mtime, so the board computes HONEST 15// recency (ALIVE/STALLED) from real data instead of age=-1 phantom labels. State is data-derived: 16// ACTIVE iff the file was touched within WSI_FRESH_SEC (genuinely in-flight), else FLOATING 17// (indexed dormant history -- honest, never a faked DONE/ACTIVE). empire = best-effort keyword 18// routing, unknown -> E-MISC (the documented catch-all, mirrors the popqueue doctrine); refined 19// later from gate markers. memory_link = the project-*.md basename (so the arc is openable). 20// 21// Sovereign: imports only nx_workstream_store + nx_seg_store + nx_framed_append + nx_ws_index_lib 22// + nx_syscalls (no gcc). Additive-only: modifies no existing organ. license_tier: ORIGINAL 23import "nx_workstream_store.nx" 24import "nx_seg_store.nx" 25import "nx_framed_append.nx" 26import "nx_ws_index_lib.nx" 27import "nx_syscalls.nx" 28const WSI_MAGIC_1048576: i64 = 1048576 29const WSI_MAGIC_1024: i64 = 1024 30 31const WSI_CAP: i64 = 8192 // max project-*.md records (426 today; ample headroom) 32const WSI_FRESH_SEC: i64 = 432000 // 5 days: a file touched within this window -> state ACTIVE 33 34// 1 if NUL-term `pre` is a prefix of NUL-term `s`. 35func wsi_has_prefix(s: *u8, pre: *u8) -> i64 { 36 var i: i64 = 0 37 while pre[i] != (0 as u8) { if s[i] != pre[i] { return 0 } i = i + 1 } 38 return 1 39} 40 41// 1 iff NUL-term `a` byte-equals NUL-term `b`. 42func wsi_id_eq(a: *u8, b: *u8) -> i64 { 43 var k: i64 = 0 44 while a[k] != (0 as u8) { if a[k] != b[k] { return 0 } k = k + 1 } 45 if b[k] != (0 as u8) { return 0 } 46 return 1 47} 48 49// copy NUL-term `s` into `out` (incl terminator); returns length. 50func wsi_setstr(out: *u8, s: *u8) -> i64 { 51 var i: i64 = 0 52 while s[i] != (0 as u8) { out[i] = s[i]; i = i + 1 } 53 out[i] = 0 as u8 54 return i 55} 56 57// strip "project-" prefix and the ".md" suffix from a filename -> a clean stream id. returns len. 58func wsi_id_from_name(name: *u8, out: *u8) -> i64 { 59 let n: i64 = wme_len(name) 60 var start: i64 = 0 61 if wsi_has_prefix(name, "project-" as *u8) == 1 { start = 8 } 62 var end: i64 = n 63 if n >= 3 { end = n - 3 } 64 var o: i64 = 0 65 var i: i64 = start 66 while i < end { out[o] = name[i]; o = o + 1; i = i + 1 } 67 out[o] = 0 as u8 68 return o 69} 70 71// 1 if NUL-term `pat` occurs as a substring of NUL-term `s` (keyword empire routing). 72func wsi_contains(s: *u8, pat: *u8) -> i64 { 73 let sl: i64 = wme_len(s) 74 if wme_find(s, sl, pat) >= 0 { return 1 } 75 return 0 76} 77 78// data-driven keyword -> empire routing for a project-*.md id. Strongest/most-specific signals 79// first; unknown -> E-MISC. Every target is a real member of ws:empires so the registry audit 80// (ws_audit_complete) stays clean. 81func wsi_empire_for(id: *u8, out: *u8) -> i64 { 82 if wsi_contains(id, "genealog" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 83 if wsi_contains(id, "lineage" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 84 if wsi_contains(id, "genesis" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 85 if wsi_contains(id, "omniforge" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 86 if wsi_contains(id, "silicon" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 87 if wsi_contains(id, "compiler" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 88 if wsi_contains(id, "nishilang" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 89 if wsi_contains(id, "kernel" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 90 if wsi_contains(id, "doctor" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 91 if wsi_contains(id, "robot" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 92 if wsi_contains(id, "pcb" as *u8) == 1 { wsi_setstr(out, "E-CORE" as *u8); return 0 } 93 if wsi_contains(id, "nishios" as *u8) == 1 { wsi_setstr(out, "E-NISHIOS" as *u8); return 0 } 94 if wsi_contains(id, "login" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 95 if wsi_contains(id, "hr-user" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 96 if wsi_contains(id, "wiki" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 97 if wsi_contains(id, "cms" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 98 if wsi_contains(id, "hosting" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 99 if wsi_contains(id, "brand" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 100 if wsi_contains(id, "andelin" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 101 if wsi_contains(id, "search" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 102 if wsi_contains(id, "publish" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 103 if wsi_contains(id, "site" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 104 if wsi_contains(id, "browser" as *u8) == 1 { wsi_setstr(out, "E-WEB" as *u8); return 0 } 105 if wsi_contains(id, "media" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 106 if wsi_contains(id, "torrent" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 107 if wsi_contains(id, "reader" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 108 if wsi_contains(id, "library" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 109 if wsi_contains(id, "gallery" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 110 if wsi_contains(id, "video" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 111 if wsi_contains(id, "image" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 112 if wsi_contains(id, "visual" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 113 if wsi_contains(id, "viz" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 114 if wsi_contains(id, "roku" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 115 if wsi_contains(id, "hevc" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 116 if wsi_contains(id, "codec" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 117 if wsi_contains(id, "gpu" as *u8) == 1 { wsi_setstr(out, "E-PLATFORM" as *u8); return 0 } 118 if wsi_contains(id, "apm" as *u8) == 1 { wsi_setstr(out, "E-DEPLOY" as *u8); return 0 } 119 if wsi_contains(id, "monitor" as *u8) == 1 { wsi_setstr(out, "E-DEPLOY" as *u8); return 0 } 120 if wsi_contains(id, "deploy" as *u8) == 1 { wsi_setstr(out, "E-DEPLOY" as *u8); return 0 } 121 if wsi_contains(id, "infra" as *u8) == 1 { wsi_setstr(out, "E-DEPLOY" as *u8); return 0 } 122 if wsi_contains(id, "vault" as *u8) == 1 { wsi_setstr(out, "E-DEPLOY" as *u8); return 0 } 123 if wsi_contains(id, "nas-" as *u8) == 1 { wsi_setstr(out, "E-DEPLOY" as *u8); return 0 } 124 if wsi_contains(id, "dns" as *u8) == 1 { wsi_setstr(out, "E-DEPLOY" as *u8); return 0 } 125 if wsi_contains(id, "host" as *u8) == 1 { wsi_setstr(out, "E-DEPLOY" as *u8); return 0 } 126 if wsi_contains(id, "research" as *u8) == 1 { wsi_setstr(out, "E-RESEARCH" as *u8); return 0 } 127 if wsi_contains(id, "manheim" as *u8) == 1 { wsi_setstr(out, "E-RESEARCH" as *u8); return 0 } 128 if wsi_contains(id, "deepmind" as *u8) == 1 { wsi_setstr(out, "E-RESEARCH" as *u8); return 0 } 129 if wsi_contains(id, "nofloat" as *u8) == 1 { wsi_setstr(out, "E-RESEARCH" as *u8); return 0 } 130 if wsi_contains(id, "model" as *u8) == 1 { wsi_setstr(out, "E-RESEARCH" as *u8); return 0 } 131 if wsi_contains(id, "analyst" as *u8) == 1 { wsi_setstr(out, "E-RESEARCH" as *u8); return 0 } 132 if wsi_contains(id, "game" as *u8) == 1 { wsi_setstr(out, "E-NGE" as *u8); return 0 } 133 if wsi_contains(id, "pets" as *u8) == 1 { wsi_setstr(out, "E-NGE" as *u8); return 0 } 134 if wsi_contains(id, "procgen" as *u8) == 1 { wsi_setstr(out, "E-NGE" as *u8); return 0 } 135 if wsi_contains(id, "pm-" as *u8) == 1 { wsi_setstr(out, "E-PM" as *u8); return 0 } 136 if wsi_contains(id, "sclass" as *u8) == 1 { wsi_setstr(out, "E-PM" as *u8); return 0 } 137 if wsi_contains(id, "workstream" as *u8) == 1 { wsi_setstr(out, "E-PM" as *u8); return 0 } 138 if wsi_contains(id, "orchestr" as *u8) == 1 { wsi_setstr(out, "E-PM" as *u8); return 0 } 139 if wsi_contains(id, "wms" as *u8) == 1 { wsi_setstr(out, "E-PM" as *u8); return 0 } 140 if wsi_contains(id, "council" as *u8) == 1 { wsi_setstr(out, "E-PM" as *u8); return 0 } 141 if wsi_contains(id, "raci" as *u8) == 1 { wsi_setstr(out, "E-PM" as *u8); return 0 } 142 wsi_setstr(out, "E-MISC" as *u8) 143 return 0 144} 145 146// build the 7-field TAB record: id \t empire \t state \t last_touched \t memory_link \t - \t - . 147func wsi_build_record(id: *u8, empire: *u8, state: *u8, mtime: i64, memlink: *u8, out: *u8) -> i64 { 148 var o: i64 = 0 149 o = wme_cat(out, o, id) 150 out[o] = 9 as u8; o = o + 1 151 o = wme_cat(out, o, empire) 152 out[o] = 9 as u8; o = o + 1 153 o = wme_cat(out, o, state) 154 out[o] = 9 as u8; o = o + 1 155 o = wme_catn(out, o, mtime) 156 out[o] = 9 as u8; o = o + 1 157 o = wme_cat(out, o, memlink) 158 out[o] = 9 as u8; o = o + 1 159 out[o] = 45 as u8; o = o + 1 // '-' code_link 160 out[o] = 9 as u8; o = o + 1 161 out[o] = 45 as u8; o = o + 1 // '-' deps 162 out[o] = 0 as u8 163 return o 164} 165 166// build the "ws:" + id key (NUL-term). returns length. 167func wsi_build_key(id: *u8, out: *u8) -> i64 { 168 out[0] = 119 as u8 // 'w' 169 out[1] = 115 as u8 // 's' 170 out[2] = 58 as u8 // ':' 171 var t: i64 = 0 172 while id[t] != (0 as u8) { out[3 + t] = id[t]; t = t + 1 } 173 out[3 + t] = 0 as u8 174 return 3 + t 175} 176 177// HANDLE-BASED idempotency compare: 1 iff open-store snapshot `h` holds `key` byte-equal to `val`. 178// (zero per-call file IO -- the scalable read path, mirrors pq_streq_h.) 179func wsi_streq_h(h: *i64, key: *u8, val: *u8) -> i64 { 180 let pq: *i64 = sys_mmap(16) as *i64 181 let lq: *i64 = sys_mmap(16) as *i64 182 if ss_hget(h, key, pq, lq) != 1 { return 0 } 183 let b: *u8 = pq[0] as *u8 184 let n: i64 = lq[0] 185 let vl: i64 = wme_len(val) 186 if n != vl { return 0 } 187 var i: i64 = 0 188 while i < n { if b[i] != val[i] { return 0 } i = i + 1 } 189 return 1 190} 191 192// merged ws:ids = existing index + every collected id not already a member, into out (NUL-term). 193func wsi_merge_ids(h: *i64, ids: *i64, nids: i64, out: *u8) -> i64 { 194 var o: i64 = 0 195 let pq: *i64 = sys_mmap(16) as *i64 196 let lq: *i64 = sys_mmap(16) as *i64 197 var cur: *u8 = "" as *u8 198 var curn: i64 = 0 199 if ss_hget(h, "ws:ids" as *u8, pq, lq) == 1 { cur = pq[0] as *u8; curn = lq[0] } 200 var t: i64 = 0 201 while t < curn { out[o] = cur[t]; o = o + 1; t = t + 1 } 202 out[o] = 0 as u8 203 var i: i64 = 0 204 while i < nids { 205 let id: *u8 = ids[i] as *u8 206 if ws_member(out, o, id) == 0 { 207 if o > 0 { out[o] = 9 as u8; o = o + 1 } 208 var j: i64 = 0 209 while id[j] != (0 as u8) { out[o] = id[j]; o = o + 1; j = j + 1 } 210 out[o] = 0 as u8 211 } 212 i = i + 1 213 } 214 return o 215} 216 217// derive a stream id from every scanned name into ids_out[] (mmap'd NUL-term ids). 218func wsi_derive_ids(names: *i64, cnt: i64, ids_out: *i64) -> i64 { 219 var i: i64 = 0 220 while i < cnt { 221 let nm: *u8 = names[i] as *u8 222 let idb: *u8 = sys_mmap(256) 223 wsi_id_from_name(nm, idb) 224 ids_out[i] = idb as i64 225 i = i + 1 226 } 227 return cnt 228} 229 230// stage ONE arc record into writer `w` against snapshot `h` (idempotent skip). Tallies 231// out4[1]=registered_new / out4[2]=skipped / out4[3]=errors. Small fn (rule 3: avoid the 232// register-pressure ceiling that mis-compiles a fat body). 233func wsi_reg_one(h: *i64, w: *i64, id: *u8, name: *u8, mtime: i64, nowsec: i64, keybuf: *u8, recbuf: *u8, empbuf: *u8, out4: *i64) -> i64 { 234 wsi_empire_for(id, empbuf) 235 wsi_build_key(id, keybuf) 236 var st: *u8 = "FLOATING" as *u8 237 if (nowsec - mtime) < WSI_FRESH_SEC { st = "ACTIVE" as *u8 } 238 let rlen: i64 = wsi_build_record(id, empbuf, st, mtime, name, recbuf) 239 if wsi_streq_h(h, keybuf, recbuf) == 1 { out4[2] = out4[2] + 1; return 0 } 240 let rc: i64 = ss_add(w, 1, keybuf, recbuf, rlen) 241 if rc < 0 { out4[3] = out4[3] + 1; return rc } 242 out4[1] = out4[1] + 1 243 return 0 244} 245 246// stage the merged ws:ids index into `w` iff it changed (idempotent). Small helper. 247func wsi_stage_ids(h: *i64, w: *i64, ids: *i64, cnt: i64) -> i64 { 248 let mids: *u8 = sys_mmap(WSI_MAGIC_1048576) 249 let mlen: i64 = wsi_merge_ids(h, ids, cnt, mids) 250 if wsi_streq_h(h, "ws:ids" as *u8, mids) == 0 { ss_add(w, 1, "ws:ids" as *u8, mids, mlen) } 251 return 0 252} 253 254// stage every per-arc ws:<id> record into `w` (skipping `omit` for the gate's neg-control). Small 255// helper -- keeps the loop body + scratch buffers OUT of the core (rule 3). 256func wsi_stage_records(h: *i64, w: *i64, names: *i64, mtimes: *i64, ids: *i64, cnt: i64, nowsec: i64, omit: *u8, out4: *i64) -> i64 { 257 let keybuf: *u8 = sys_mmap(512) 258 let recbuf: *u8 = sys_mmap(WSI_MAGIC_1024) 259 let empbuf: *u8 = sys_mmap(64) 260 var has_omit: i64 = 0 261 if omit as i64 != 0 { has_omit = 1 } 262 var i: i64 = 0 263 while i < cnt { 264 let id: *u8 = ids[i] as *u8 265 let nm: *u8 = names[i] as *u8 266 var skip_this: i64 = 0 267 if has_omit == 1 { if wsi_id_eq(id, omit) == 1 { skip_this = 1 } } 268 if skip_this == 0 { wsi_reg_one(h, w, id, nm, mtimes[i], nowsec, keybuf, recbuf, empbuf, out4) } 269 i = i + 1 270 } 271 return 0 272} 273 274// THE CORE: scan every project-*.md in `memdir` and register each into the registry under 275// `prefix` in ONE snapshot / ONE writer / ONE commit (256-cap safe, idempotent). `omit` = a 276// NUL-term id to deliberately skip (or 0) -- the gate's negative control. Counters via out4[]: 277// out4[0]=scanned out4[1]=registered_new out4[2]=skipped out4[3]=errors 278// return: 0 ok, <0 hard error. 279func wsi_ingest_omit_p(prefix: *u8, memdir: *u8, omit: *u8, out4: *i64) -> i64 { 280 let names: *i64 = sys_mmap(8 * WSI_CAP) as *i64 281 let mtimes: *i64 = sys_mmap(8 * WSI_CAP) as *i64 282 let cnt: i64 = wme_scan(memdir, names, mtimes, WSI_CAP) 283 out4[0] = 0; out4[1] = 0; out4[2] = 0; out4[3] = 0 284 if cnt < 0 { out4[3] = 1; return 0 - 1 } 285 out4[0] = cnt 286 let ids: *i64 = sys_mmap(8 * WSI_CAP) as *i64 287 wsi_derive_ids(names, cnt, ids) 288 let nowsec: i64 = sys_now_realtime_sec() 289 let h: *i64 = ss_open(prefix) 290 let w: *i64 = ss_begin() 291 wsi_stage_ids(h, w, ids, cnt) 292 wsi_stage_records(h, w, names, mtimes, ids, cnt, nowsec, omit, out4) 293 if w[1] > 0 { 294 let segid: i64 = ws_seg_next(prefix) 295 let crc: i64 = ss_commit(prefix, w, segid) 296 if crc != 0 { out4[3] = out4[3] + 1; return crc } 297 } 298 if out4[3] > 0 { return 0 - 2 } 299 return 0 300} 301 302// production convenience: ingest into the live WS_PREFIX, omit nothing. 303func wsi_ingest(memdir: *u8, out4: *i64) -> i64 { 304 return wsi_ingest_omit_p(WS_PREFIX, memdir, 0 as *u8, out4) 305} 306 307// NOTE: pure LIB (no main) so the gate can import it. The runnable driver is nx_ws_ingest_run.nx.