nx_orchestrate.nx source

1// nx_orchestrate.nx -- auto lane (eats debt seq163). WAIT-FOR-OPENING QUEUE: sessions QUEUE actions blocked 2// by a sibling's in-flight work instead of colliding. Each knowledge/store/deployq- row = 3// id | title | precondition | status | planid | note 4// precondition = a CHECKABLE predicate, now COMPOUND-capable (v2, 2026-07-20 -- for safe mgmt deploys that 5// must fire only when the fix is STAGED *and* mgmt is CALM): 6// exists:<path> | contains:<path>:<needle> -- single, OR 7// <pred> && <pred> -- AND of two (BOTH must hold to fire) 8// (e.g. exists:nx_mgmt_api.elf.new && contains:knowledge/status/api_contract.log:VERDICT=GREEN). 9// This POLLER (cron */5, non-blocking) evaluates every `queued` row: satisfied -> FIRE the row's plan via 10// nx_plan_run (reusing its allowlist+pinned-arg security WHOLESALE -- an action is a pre-seeded plan- 11// workflow, never arbitrary exec) and re-put the row status=fired-rc<N>; unsatisfied -> WAIT (logged, no 12// dead thread, no clobber). SOTA: Temporal durable await-condition, sovereign. Never-brick: fires only plans. 13// nx_orchestrate run [qprefix] [outlog] 14// ENVELOPE: 64 rows/pass, precond file window 256KB, fields 800B; declared in output. 15// exit: 0 ran (fired/waiting/skipped counted) | 4 queue unreadable | 2 usage. 16// license_tier: ORIGINAL expect_exit: 0 17import "nx_tool_run.nx" 18const K_MAGIC_30000: i64 = 30000 19const K_MAGIC_262144: i64 = 262144 20const K_MAGIC_262160: i64 = 262160 21const K_MAGIC_65536: i64 = 65536 22const K_MAGIC_32768: i64 = 32768 23const K_MAGIC_65535: i64 = 65535 24 25func oc_w(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 } 26func oc_b(rep: *u8, pos: i64, s: *u8) -> i64 { var p: i64 = pos; var i: i64 = 0; while s[i] != (0 as u8) { if p < K_MAGIC_30000 { rep[p] = s[i]; p = p + 1 } i = i + 1 } return p } 27func oc_bn(rep: *u8, pos: i64, v: i64) -> i64 { var p: i64 = pos; var m: i64 = v; if m < 0 { if p < K_MAGIC_30000 { rep[p] = 45 as u8; p = p + 1 } m = 0 - m } let t: *u8 = sys_mmap(28); var k: i64 = 0; if m == 0 { t[0] = 48 as u8; 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 { if p < K_MAGIC_30000 { rep[p] = t[k - 1 - i]; p = p + 1 } i = i + 1 } return p } 28func oc_reads(path: *u8, buf: *u8, cap: i64) -> i64 { 29 let fd: i64 = sys_openat_rd(path) 30 if fd < 0 { return 0 - 1 } 31 var n: i64 = 0 32 var go: i64 = 1 33 while go == 1 { let r: i64 = sys_read(fd, (buf as i64 + n) as *u8, cap - n); if r <= 0 { go = 0 } else { n = n + r } if n >= cap { go = 0 } } 34 sys_close(fd) 35 return n 36} 37func oc_has(buf: *u8, s: i64, e: i64, pat: *u8) -> i64 { 38 var pl: i64 = 0 39 while pat[pl] != (0 as u8) { pl = pl + 1 } 40 if pl == 0 { return 0 } 41 var i: i64 = s 42 var hit: i64 = 0 43 while i + pl <= e { 44 var k: i64 = 0 45 var m: i64 = 1 46 while k < pl { if buf[i + k] != pat[k] { m = 0; k = pl } else { k = k + 1 } } 47 if m == 1 { hit = 1; i = e } else { i = i + 1 } 48 } 49 return hit 50} 51func oc_cpz(dst: *u8, cap: i64, buf: *u8, s: i64, e: i64) -> i64 { 52 var d: i64 = 0 53 var i: i64 = s 54 while i < e { if d < cap - 1 { dst[d] = buf[i]; d = d + 1 } i = i + 1 } 55 dst[d] = 0 as u8 56 return d 57} 58func oc_pre(s: *u8, pre: *u8) -> i64 { 59 var i: i64 = 0 60 while pre[i] != (0 as u8) { if s[i] != pre[i] { return 0 - 1 } i = i + 1 } 61 return i 62} 63// evaluate ONE predicate (null-term): 1 satisfied, 0 unsatisfied, -2 malformed. fbuf = scratch 256KB. 64func oc_eval(pred: *u8, fpath: *u8, fneedle: *u8, fbuf: *u8) -> i64 { 65 let ex: i64 = oc_pre(pred, "exists:" as *u8) 66 if ex > 0 { 67 var d2: i64 = 0 68 var s2: i64 = ex 69 while pred[s2] != (0 as u8) { if d2 < 598 { fpath[d2] = pred[s2]; d2 = d2 + 1 } s2 = s2 + 1 } 70 fpath[d2] = 0 as u8 71 let tfd: i64 = sys_openat_rd(fpath) 72 if tfd >= 0 { sys_close(tfd); return 1 } 73 return 0 74 } 75 let co: i64 = oc_pre(pred, "contains:" as *u8) 76 if co > 0 { 77 var d3: i64 = 0 78 var s3: i64 = co 79 var stop: i64 = 0 80 while stop == 0 { let c: i64 = pred[s3]; if c == 0 { stop = 1 } else { if c == 58 { stop = 2 } else { if d3 < 598 { fpath[d3] = c as u8; d3 = d3 + 1 } s3 = s3 + 1 } } } 81 fpath[d3] = 0 as u8 82 if stop != 2 { return 0 - 2 } 83 var d4: i64 = 0 84 var s4: i64 = s3 + 1 85 while pred[s4] != (0 as u8) { if d4 < 598 { fneedle[d4] = pred[s4]; d4 = d4 + 1 } s4 = s4 + 1 } 86 fneedle[d4] = 0 as u8 87 let fn2: i64 = oc_reads(fpath, fbuf, K_MAGIC_262144) 88 if fn2 < 0 { return 0 } 89 return oc_has(fbuf, 0, fn2, fneedle) 90 } 91 return 0 - 2 92} 93// evaluate a (possibly compound A && B) precondition string. 94func oc_eval_pre(fpre: *u8, fpath: *u8, fneedle: *u8, fbuf: *u8, lbuf: *u8, rbuf: *u8) -> i64 { 95 // find "&&" 96 var amp: i64 = 0 - 1 97 var i: i64 = 0 98 while fpre[i] != (0 as u8) { if fpre[i] == (38 as u8) { if fpre[i+1] == (38 as u8) { amp = i; } } if amp >= 0 { i = i } i = i + 1 } 99 if amp < 0 { return oc_eval(fpre, fpath, fneedle, fbuf) } 100 // left = [0, amp) trimmed of trailing spaces; right = after amp+2, skipping leading spaces 101 var le: i64 = amp 102 while le > 0 { if fpre[le-1] == (32 as u8) { le = le - 1 } else { let z: i64 = le; le = 0 - 2 - z } } 103 if le < 0 { le = 0 - 2 - le } 104 var d: i64 = 0 105 var j: i64 = 0 106 while j < le { if d < 700 { lbuf[d] = fpre[j]; d = d + 1 } j = j + 1 } 107 lbuf[d] = 0 as u8 108 var rs: i64 = amp + 2 109 var sk: i64 = 1 110 while sk == 1 { if fpre[rs] == (32 as u8) { rs = rs + 1 } else { sk = 0 } } 111 var d2: i64 = 0 112 while fpre[rs] != (0 as u8) { if d2 < 700 { rbuf[d2] = fpre[rs]; d2 = d2 + 1 } rs = rs + 1 } 113 rbuf[d2] = 0 as u8 114 let l: i64 = oc_eval(lbuf, fpath, fneedle, fbuf) 115 if l == (0 - 2) { return 0 - 2 } 116 let r: i64 = oc_eval(rbuf, fpath, fneedle, fbuf) 117 if r == (0 - 2) { return 0 - 2 } 118 if l == 1 { if r == 1 { return 1 } } 119 return 0 120} 121func main(argc: i64, argv: *i64) -> i64 { 122 if argc < 2 { oc_w(2, "usage: nx_orchestrate run [qprefix] [outlog]\n" as *u8); sys_exit(2); return 2 } 123 var qprefix: *u8 = "knowledge/store/deployq-" as *u8 124 var outlog: *u8 = "knowledge/status/orchestrate.log" as *u8 125 if argc >= 3 { qprefix = argv[2] as *u8 } 126 if argc >= 4 { outlog = argv[3] as *u8 } 127 let SP: *u8 = "/volume1/homes/elderwesto/nishihost/nx_store_put.elf" as *u8 128 let PR: *u8 = "/volume1/homes/elderwesto/nishihost/nx_plan_run.elf" as *u8 129 let qcap: i64 = K_MAGIC_262144 130 let qbuf: *u8 = sys_mmap(qcap + 16) 131 let olen: *i64 = sys_mmap(16) as *i64 132 let av: *i64 = sys_mmap(16 * 8) as *i64 133 av[0] = SP as i64 134 av[1] = qprefix as i64 135 av[2] = "load" as *u8 as i64 136 av[3] = 0 137 let lrc: i64 = tr_run_capture(SP, av, qbuf, qcap, olen) 138 let qn: i64 = olen[0] 139 if lrc == 127 { oc_w(1, "ORCHESTRATE verdict=QUEUE-UNREADABLE exec-127\n" as *u8); sys_exit(4); return 4 } 140 if lrc < 0 { oc_w(1, "ORCHESTRATE verdict=QUEUE-UNREADABLE harness\n" as *u8); sys_exit(4); return 4 } 141 let fid: *u8 = sys_mmap(128) 142 let ftitle: *u8 = sys_mmap(800) 143 let fpre: *u8 = sys_mmap(800) 144 let fstat: *u8 = sys_mmap(64) 145 let fplan: *u8 = sys_mmap(128) 146 let fnote: *u8 = sys_mmap(800) 147 let fpath: *u8 = sys_mmap(700) 148 let fneedle: *u8 = sys_mmap(700) 149 let lbuf: *u8 = sys_mmap(800) 150 let rbuf: *u8 = sys_mmap(800) 151 let stat2: *u8 = sys_mmap(64) 152 let fbuf: *u8 = sys_mmap(K_MAGIC_262160) 153 let cbuf: *u8 = sys_mmap(K_MAGIC_65536) 154 let rep: *u8 = sys_mmap(K_MAGIC_32768) 155 var p: i64 = 0 156 p = oc_b(rep, p, "ORCHESTRATE run queue_bytes=" as *u8); p = oc_bn(rep, p, qn) 157 p = oc_b(rep, p, " (v2 compound-AND; caps 64 rows/pass, 256KB precond; predicates exists:/contains: [&&]; fire=nx_plan_run)\n" as *u8) 158 var fired: i64 = 0 159 var waiting: i64 = 0 160 var skipped: i64 = 0 161 var malformed: i64 = 0 162 var rows: i64 = 0 163 var i: i64 = 0 164 var ls: i64 = 0 165 while i <= qn { 166 var isend: i64 = 0 167 if i == qn { isend = 1 } else { if qbuf[i] == (10 as u8) { isend = 1 } } 168 if isend == 1 { 169 if i > ls { if rows < 64 { 170 rows = rows + 1 171 let fs2: *i64 = sys_mmap(64) as *i64 172 let fe2: *i64 = sys_mmap(64) as *i64 173 var nf: i64 = 1 174 fs2[0] = ls 175 var q: i64 = ls 176 while q < i { if qbuf[q] == (9 as u8) { if nf < 6 { fe2[nf-1] = q; fs2[nf] = q + 1; nf = nf + 1 } } q = q + 1 } 177 fe2[nf-1] = i 178 if nf >= 4 { 179 oc_cpz(fid, 128, qbuf, fs2[0], fe2[0]) 180 oc_cpz(ftitle, 800, qbuf, fs2[1], fe2[1]) 181 oc_cpz(fpre, 800, qbuf, fs2[2], fe2[2]) 182 oc_cpz(fstat, 64, qbuf, fs2[3], fe2[3]) 183 fplan[0] = 0 as u8 184 fnote[0] = 0 as u8 185 if nf >= 5 { oc_cpz(fplan, 128, qbuf, fs2[4], fe2[4]) } 186 if nf >= 6 { oc_cpz(fnote, 800, qbuf, fs2[5], fe2[5]) } 187 var isq: i64 = 0 188 if fstat[0] == 113 { if fstat[1] == 117 { if fstat[2] == 101 { if fstat[3] == 117 { if fstat[4] == 101 { if fstat[5] == 100 { if fstat[6] == (0 as u8) { isq = 1 } } } } } } } 189 if isq == 0 { skipped = skipped + 1 } else { 190 let sat: i64 = oc_eval_pre(fpre, fpath, fneedle, fbuf, lbuf, rbuf) 191 if sat == (0 - 2) { 192 malformed = malformed + 1 193 p = oc_b(rep, p, "MALFORMED-PRED id=" as *u8); p = oc_b(rep, p, fid); p = oc_b(rep, p, "\n" as *u8) 194 } else { if sat == 1 { 195 if fplan[0] == (0 as u8) { 196 malformed = malformed + 1 197 p = oc_b(rep, p, "NO-PLANID id=" as *u8); p = oc_b(rep, p, fid); p = oc_b(rep, p, "\n" as *u8) 198 } else { 199 let av2: *i64 = sys_mmap(16 * 8) as *i64 200 av2[0] = PR as i64 201 av2[1] = fplan as i64 202 av2[2] = 0 203 let olen2: *i64 = sys_mmap(16) as *i64 204 let prc: i64 = tr_run_capture(PR, av2, cbuf, K_MAGIC_65535, olen2) 205 fired = fired + 1 206 var sp2: i64 = 0 207 let sfx: *u8 = "fired-rc" as *u8 208 while sfx[sp2] != (0 as u8) { stat2[sp2] = sfx[sp2]; sp2 = sp2 + 1 } 209 var m2: i64 = prc 210 if m2 < 0 { m2 = 99 } 211 if m2 >= 100 { m2 = 99 } 212 if m2 >= 10 { stat2[sp2] = (48 + (m2 / 10)) as u8; sp2 = sp2 + 1 } 213 stat2[sp2] = (48 + (m2 % 10)) as u8 214 stat2[sp2 + 1] = 0 as u8 215 let av3: *i64 = sys_mmap(16 * 8) as *i64 216 av3[0] = SP as i64 217 av3[1] = qprefix as i64 218 av3[2] = "put" as *u8 as i64 219 av3[3] = "nx_orchestrate" as *u8 as i64 220 av3[4] = fid as i64 221 av3[5] = ftitle as i64 222 av3[6] = fpre as i64 223 av3[7] = stat2 as i64 224 av3[8] = fplan as i64 225 av3[9] = fnote as i64 226 av3[10] = 0 227 let olen3: *i64 = sys_mmap(16) as *i64 228 tr_run_capture(SP, av3, cbuf, K_MAGIC_65535, olen3) 229 p = oc_b(rep, p, "FIRED id=" as *u8); p = oc_b(rep, p, fid) 230 p = oc_b(rep, p, " plan=" as *u8); p = oc_b(rep, p, fplan) 231 p = oc_b(rep, p, " rc=" as *u8); p = oc_bn(rep, p, prc) 232 p = oc_b(rep, p, "\n" as *u8) 233 } 234 } else { 235 waiting = waiting + 1 236 p = oc_b(rep, p, "WAITING id=" as *u8); p = oc_b(rep, p, fid) 237 p = oc_b(rep, p, " precond=" as *u8); p = oc_b(rep, p, fpre) 238 p = oc_b(rep, p, "\n" as *u8) 239 } } 240 } 241 } 242 } } 243 ls = i + 1 244 } 245 i = i + 1 246 } 247 p = oc_b(rep, p, "SUMMARY rows=" as *u8); p = oc_bn(rep, p, rows) 248 p = oc_b(rep, p, " fired=" as *u8); p = oc_bn(rep, p, fired) 249 p = oc_b(rep, p, " waiting=" as *u8); p = oc_bn(rep, p, waiting) 250 p = oc_b(rep, p, " skipped=" as *u8); p = oc_bn(rep, p, skipped) 251 p = oc_b(rep, p, " malformed=" as *u8); p = oc_bn(rep, p, malformed) 252 p = oc_b(rep, p, "\n" as *u8) 253 sys_write(1, rep, p) 254 let lfd: i64 = sys_openat_append(outlog, 420) 255 if lfd >= 0 { sys_write(lfd, rep, p); sys_close(lfd) } 256 let tfd2: i64 = sys_openat_wr("knowledge/status/orchestrate.out.tmp" as *u8, 420) 257 if tfd2 >= 0 { sys_write(tfd2, rep, p); sys_close(tfd2); sys_renameat("knowledge/status/orchestrate.out.tmp" as *u8, "knowledge/status/orchestrate.out" as *u8) } 258 sys_exit(0) 259 return 0 260}