nx_queue_load.nx source

1// nx_queue_load.nx -- IM-Q1: the PROGRAM'S OWN COORDINATION DATA becomes 2// first-class records on the sovereign store (operator 2026-06-11: "own our 3// information management -- a tsv file is not nishi; bottom up hardware rung 4// and each rung nishi"). The seg-store substrate is already hardware-up 5// (rungs 1-9, beats FTS5); this rung puts the QUEUE on it: 6// assignment_queue.tsv rows -> key "aq:"+<row-id>, value = canonical record 7// {kind: queue-row, id, row}. SAME key, NEW VERSION on every change -- 8// `nx_store_query hist aq:X-Q-003` is the row's full status time-travel, 9// something the bare TSV can never answer. Idempotent: unchanged row = 10// identical canonical bytes = dup_instore, never re-added. 11// sponsor_decisions.tsv / planning_prefs.tsv / callouts.tsv lines -> 12// content-CID line records (sd:/pp:/co: prefixes, nx_log_load pattern) -- 13// the sponsor's decisions + the training corpus gain provenance lineage. 14// The TSV files stay as the WRITE surface this rung (API stability); IM-Q2 15// (filed) moves writers store-first and derives the TSV as a render. 16// argv[1]=store prefix override, argv[2]=queue path override (gates). 17// Evidence: QUEUE-STORE row -> stdout + knowledge/status/infomgmt_store.log. 18// license_tier: ORIGINAL 19 20import "nx_syscalls.nx" 21import "nx_canon_cid.nx" 22import "nx_seg_store.nx" 23const QL_MAGIC_3800: i64 = 3800 24const QL_MAGIC_4096: i64 = 4096 25const QL_MAGIC_8192: i64 = 8192 26 27func ql_p(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 28func ql_fp(fd: i64, s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(fd,s,n); return 0 } 29func ql_fn(fd: i64, v: i64) -> i64 { let bb: *u8=sys_mmap(28); var m: i64=v; if m<0{m=0-m; sys_write(fd,"-" as *u8,1)}; let t: *u8=sys_mmap(28); var k: i64=0; if m==0{t[0]=48;k=1}; while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1}; var i: i64=0; while i<k{bb[i]=t[k-1-i];i=i+1}; sys_write(fd,bb,k); return 0 } 30 31func ql_memeq(a: *u8, b: *u8, n: i64) -> i64 { 32 var i: i64 = 0 33 while i < n { 34 if a[i] != b[i] { return 0 } 35 i = i + 1 36 } 37 return 1 38} 39 40// st: st[0]=writer st[1]=wdocs st[2]=segs st[3]=added st[4]=dup st[5]=scanned 41// st[6]=abort-rc st[7]=open handle 42const QL_CHUNK: i64 = 64 43 44func ql_flush(prefix: *u8, st: *i64) -> i64 { 45 if st[1] == 0 { return 0 } 46 let rc: i64 = ss_commit(prefix, st[0] as *i64, sys_now_us()) 47 if rc != 0 { st[6] = rc; return rc } 48 st[0] = ss_begin() as i64 49 st[1] = 0 50 st[2] = st[2] + 1 51 return 0 52} 53 54// add record if the store's latest version differs (or key absent); 0/neg 55func ql_put(prefix: *u8, st: *i64, kbuf: *u8, enc: *u8, el: i64) -> i64 { 56 let pp: *i64 = sys_mmap(16) as *i64 57 let ll: *i64 = sys_mmap(16) as *i64 58 var have: i64 = 0 59 let g: i64 = ss_hget(st[7] as *i64, kbuf, pp, ll) 60 if g == 1 { if ll[0] == el { if ql_memeq(pp[0] as *u8, enc, el) == 1 { have = 1 } } } 61 if have == 1 { st[4] = st[4] + 1; return 0 } 62 if ss_add(st[0] as *i64, 1, kbuf, enc, el) != 0 { st[6] = 0 - 98; return 0 - 3 } 63 st[3] = st[3] + 1 64 st[1] = st[1] + 1 65 if st[1] >= QL_CHUNK { if ql_flush(prefix, st) != 0 { return 0 - 4 } } 66 return 0 67} 68 69// one queue row line [i,lend) -> aq:<id> versioned record 70func ql_row(prefix: *u8, st: *i64, b: *u8, i: i64, lend: i64) -> i64 { 71 let llen: i64 = lend - i 72 if llen <= 0 { return 0 } 73 if b[i] == (35 as u8) { return 0 } 74 if llen > QL_MAGIC_3800 { return 0 } 75 // id = field 0 (to first tab); no tab = not a row 76 var idl: i64 = 0 77 var go: i64 = 1 78 while go == 1 { 79 if i + idl >= lend { return 0 } else { 80 if b[i + idl] == (9 as u8) { go = 0 } else { 81 idl = idl + 1 82 if idl > 40 { return 0 } 83 } 84 } 85 } 86 if idl == 0 { return 0 } 87 st[5] = st[5] + 1 88 let idb: *u8 = sys_mmap(64) 89 var t: i64 = 0 90 while t < idl { idb[t] = b[i + t]; t = t + 1 } 91 idb[idl] = 0 as u8 92 let lbuf: *u8 = sys_mmap(QL_MAGIC_4096) 93 t = 0 94 while t < llen { lbuf[t] = b[i + t]; t = t + 1 } 95 lbuf[llen] = 0 as u8 96 let keys: *i64 = sys_mmap(8 * 4) as *i64 97 let vals: *i64 = sys_mmap(8 * 4) as *i64 98 keys[0] = "kind" as *u8 as i64 99 keys[1] = "id" as *u8 as i64 100 keys[2] = "row" as *u8 as i64 101 vals[0] = "queue-row" as *u8 as i64 102 vals[1] = idb as i64 103 vals[2] = lbuf as i64 104 let enc: *u8 = sys_mmap(QL_MAGIC_8192) 105 let el: i64 = canon_encode(keys, vals, 3, enc) 106 let kbuf: *u8 = sys_mmap(128) 107 var ko: i64 = 0 108 ko = ss_cat(kbuf, ko, "aq:" as *u8) 109 ko = ss_cat(kbuf, ko, idb) 110 kbuf[ko] = 0 as u8 111 return ql_put(prefix, st, kbuf, enc, el) 112} 113 114func ql_load_queue(prefix: *u8, path: *u8, st: *i64) -> i64 { 115 let szp: *i64 = sys_mmap(16) as *i64 116 let b: *u8 = ss_readall(path, szp) 117 let sz: i64 = szp[0] 118 if sz <= 0 { return 0 - 1 } 119 var i: i64 = 0 120 while i < sz { 121 var e: i64 = i 122 var go: i64 = 1 123 while go == 1 { 124 if e >= sz { go = 0 } else { 125 if b[e] == (10 as u8) { go = 0 } else { e = e + 1 } 126 } 127 } 128 if ql_row(prefix, st, b, i, e) < 0 { return 0 - 2 } 129 i = e + 1 130 } 131 return 0 132} 133 134// line-CID records for a sponsor-surface file (nx_log_load pattern) 135func ql_load_lines(prefix: *u8, path: *u8, tag: *u8, st: *i64) -> i64 { 136 let szp: *i64 = sys_mmap(16) as *i64 137 let b: *u8 = ss_readall(path, szp) 138 let sz: i64 = szp[0] 139 if sz <= 0 { return 0 } 140 let keys: *i64 = sys_mmap(8 * 4) as *i64 141 let vals: *i64 = sys_mmap(8 * 4) as *i64 142 keys[0] = "kind" as *u8 as i64 143 keys[1] = "log" as *u8 as i64 144 keys[2] = "line" as *u8 as i64 145 vals[0] = "sponsor-surface-line" as *u8 as i64 146 vals[1] = tag as i64 147 let lbuf: *u8 = sys_mmap(QL_MAGIC_4096) 148 let enc: *u8 = sys_mmap(QL_MAGIC_8192) 149 let cid: *u8 = sys_mmap(96) 150 let kbuf: *u8 = sys_mmap(160) 151 var i: i64 = 0 152 while i < sz { 153 var e: i64 = i 154 var go: i64 = 1 155 while go == 1 { 156 if e >= sz { go = 0 } else { 157 if b[e] == (10 as u8) { go = 0 } else { e = e + 1 } 158 } 159 } 160 let llen: i64 = e - i 161 var use: i64 = 1 162 if llen <= 0 { use = 0 } 163 if llen > QL_MAGIC_3800 { use = 0 } 164 if use == 1 { if b[i] == (35 as u8) { use = 0 } } 165 if use == 1 { 166 st[5] = st[5] + 1 167 var t: i64 = 0 168 while t < llen { lbuf[t] = b[i + t]; t = t + 1 } 169 lbuf[llen] = 0 as u8 170 vals[2] = lbuf as i64 171 let el: i64 = canon_encode(keys, vals, 3, enc) 172 cid_of(enc, el, cid) 173 var ko: i64 = 0 174 ko = ss_cat(kbuf, ko, tag) 175 ko = ss_cat(kbuf, ko, ":" as *u8) 176 ko = ss_cat(kbuf, ko, cid) 177 kbuf[ko] = 0 as u8 178 if ql_put(prefix, st, kbuf, enc, el) < 0 { return 0 - 2 } 179 } 180 i = e + 1 181 } 182 return 0 183} 184 185func main(argc: i64, argv: *i64) -> i64 { 186 var prefix: *u8 = "knowledge/store/im-" as *u8 187 var qpath: *u8 = "knowledge/registry/assignment_queue.tsv" as *u8 188 if argc >= 2 { prefix = argv[1] as *u8 } 189 if argc >= 3 { qpath = argv[2] as *u8 } 190 let logfd: i64 = sys_openat_append("knowledge/status/infomgmt_store.log" as *u8, 0x1a4) 191 192 let st: *i64 = sys_mmap(8 * 16) as *i64 193 st[0] = ss_begin() as i64 194 st[7] = ss_open(prefix) as i64 195 196 if ql_load_queue(prefix, qpath, st) != 0 { 197 ql_p("QUEUE-STORE FAILED rc-inner=" as *u8); ql_fn(1, st[6]); ql_p(" -- fail loud\n" as *u8) 198 sys_exit(1) 199 } 200 // sponsor surfaces ride the same pass (real paths only; scratch gates 201 // pass a scratch prefix so these dedup against it harmlessly) 202 ql_load_lines(prefix, "knowledge/registry/sponsor_decisions.tsv" as *u8, "sd" as *u8, st) 203 ql_load_lines(prefix, "knowledge/registry/planning_prefs.tsv" as *u8, "pp" as *u8, st) 204 ql_load_lines(prefix, "knowledge/registry/callouts.tsv" as *u8, "co" as *u8, st) 205 if st[6] != 0 { 206 ql_p("QUEUE-STORE FAILED late rc=" as *u8); ql_fn(1, st[6]); ql_p("\n" as *u8) 207 sys_exit(1) 208 } 209 if ql_flush(prefix, st) != 0 { 210 ql_p("QUEUE-STORE final commit FAILED -- fail loud\n" as *u8) 211 sys_exit(1) 212 } 213 214 // probe: the queue is QUERYABLE on the team's own substrate 215 let h2: *i64 = ss_open(prefix) 216 let pp2: *i64 = sys_mmap(16) as *i64 217 let ll2: *i64 = sys_mmap(16) as *i64 218 let probe: i64 = ss_hget(h2, "aq:X-Q-003" as *u8, pp2, ll2) 219 220 var fdi: i64 = 0 221 while fdi < 2 { 222 var fd: i64 = 1 223 if fdi == 1 { fd = logfd } 224 if fd > 0 { 225 ql_fp(fd, "QUEUE-STORE scanned=" as *u8); ql_fn(fd, st[5]) 226 ql_fp(fd, " added=" as *u8); ql_fn(fd, st[3]) 227 ql_fp(fd, " dup_instore=" as *u8); ql_fn(fd, st[4]) 228 ql_fp(fd, " segments=" as *u8); ql_fn(fd, st[2]) 229 ql_fp(fd, " probe_aq=" as *u8); ql_fn(fd, probe) 230 ql_fp(fd, " epoch=" as *u8); ql_fn(fd, sys_now_realtime_sec()) 231 ql_fp(fd, " verdict=GREEN\n" as *u8) 232 } 233 fdi = fdi + 1 234 } 235 if logfd > 0 { sys_close(logfd) } 236 sys_exit(0) 237 return 0 238}