nx_eval_queue.nx source
↩ module page · 225 lines · 12515 B
1// nx_eval_queue.nx -- THE REPORT -> LOG-NEXT BRIDGE of the autonomy loop: turn an evaluator/scanner finding
2// (a rung gap, an anti-pattern hit, an ORPHANED skeleton) into a WMS WORK-SIGNAL queued for completion.
3//
4// Operator's loop: DO -> TEST+EVIDENCE -> REPORT(gaps) -> RESEARCH -> GROW -> LOG-NEXT -> REPEAT. nx_loop_census
5// measured the "QUEUE" link as the gap (the live conductor self-ticks DO+TEST+GATE+LEARN but does not yet pull
6// dynamic work). THIS is that link, built composably: nx_rung_eval / nx_antipattern_catalog find a gap -> this
7// emits it as a work-item the conductor/WMS can pull -> DO. So a detected flaw becomes queued work AUTOMATICALLY,
8// and (Cardinal 25) an orphaned skeleton is ROUTED TO COMPLETION, never deleted.
9//
10// SAFE BY CONSTRUCTION (Cardinal 13 additive): writes ONLY to a DEDICATED, ISOLATED namespace
11// (knowledge/store/eval-gaps-) over the SAME ss_* substrate the WMS uses -- the live 1265-stream production
12// board (prefix "knowledge/store/ws-") is a DIFFERENT store, never opened/read/written here. Idempotent
13// (re-emit of an identical gap re-states the same key, no dup) with fresh-max+1 segids (the "never recommit a
14// segid" discipline). The record is the EXACT WMS TAB contract so a production daemon can adopt these unchanged.
15//
16// COMPOSES nx_seg_store directly (Cardinal 15) -- mirrors the proven nx_asset_signals as_emit pattern for CODE
17// gaps instead of asset signals. Each work-item carries its RUNG (family-tree anchor -- ties the gap to where
18// it lives in the dependency-graph lineage). license_tier: ORIGINAL
19// genealogy_id: operator-2026-07-03-ultimate-loop + nx_asset_signals-wms-contract
20// lineage_id: nishi_eval_queue_v1 expect_exit: 0
21import "nx_syscalls.nx"
22import "nx_itoa_lib.nx" // shared MSB-first emitter (zero-alloc)
23import "nx_seg_store.nx"
24const K_MAGIC_65536: i64 = 65536
25const K_MAGIC_2048: i64 = 2048
26
27func EQ_PREFIX() -> *u8 { return "knowledge/store/eval-gaps-\x00" as *u8 }
28
29func eq_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
30// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
31// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
32// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
33// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
34func eq_putn(v: i64) -> i64 { nxi_out(v); return 0 }
35func eq_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
36func eq_cat(dst: *u8, off: i64, src: *u8) -> i64 { var i: i64 = 0; while src[i] != (0 as u8) { dst[off + i] = src[i]; i = i + 1 } dst[off + i] = 0 as u8; return off + i }
37
38// next segment id under `prefix` = the ACTUAL fresh max+1 (ss_next_segid, UNCAPPED; mirrors as_seg_next).
39// the old 1+capped-count only APPROXIMATED max+1 and clobbered the cap segment past the cap. Empty stays 1-based.
40func eq_seg_next(prefix: *u8) -> i64 {
41 let n: i64 = ss_next_segid(prefix)
42 if n < 1 { return 1 }
43 return n
44}
45
46// idempotent record write (skip-if-byte-identical, else additive new version). 0=new / 1=unchanged / <0 err.
47func eq_put(prefix: *u8, key: *u8, val: *u8) -> i64 {
48 let pq: *i64 = sys_mmap(16) as *i64
49 let lq: *i64 = sys_mmap(16) as *i64
50 let vl: i64 = eq_len(val)
51 if ss_get(prefix, key, pq, lq) == 1 {
52 if lq[0] == vl {
53 let b: *u8 = pq[0] as *u8
54 var same: i64 = 1; var i: i64 = 0
55 while i < vl { if b[i] != val[i] { same = 0; i = vl } else { i = i + 1 } }
56 if same == 1 { return 1 }
57 }
58 }
59 let w: *i64 = ss_begin()
60 ss_add(w, 1, key, val, vl)
61 let segid: i64 = eq_seg_next(prefix)
62 return ss_commit(prefix, w, segid)
63}
64
65// whole-field TAB-member test for the ws:ids dedupe (a key must not match a longer field).
66func eq_member(list: *u8, llen: i64, tok: *u8) -> i64 {
67 let tl: i64 = eq_len(tok)
68 var i: i64 = 0
69 while i <= llen - tl {
70 var m: i64 = 0; var ok: i64 = 1
71 while m < tl { if list[i + m] != tok[m] { ok = 0; m = tl } else { m = m + 1 } }
72 if ok == 1 {
73 var lb: i64 = 1; if i > 0 { if list[i - 1] != (9 as u8) { lb = 0 } }
74 var rb: i64 = 1; let after: i64 = i + tl; if after < llen { if list[after] != (9 as u8) { rb = 0 } }
75 if lb == 1 { if rb == 1 { return 1 } }
76 }
77 i = i + 1
78 }
79 return 0
80}
81
82// EMIT ONE GAP as a queued work-item. key="gap:"+file+":"+code (idempotent). Record (WMS TAB contract):
83// field0=code field1=file field2=detail field3=OPEN field4=eval field5=rung(lineage anchor)
84// Also extends the ws:ids index so the queue is enumerable. Returns 0/1 (put result) or <0 on error.
85func nx_eval_queue_emit(rung: *u8, file: *u8, code: *u8, detail: *u8) -> i64 {
86 let prefix: *u8 = EQ_PREFIX()
87 let key: *u8 = sys_mmap(eq_len(file) + eq_len(code) + 16)
88 var ko: i64 = eq_cat(key, 0, "gap:\x00" as *u8)
89 ko = eq_cat(key, ko, file); ko = eq_cat(key, ko, ":\x00" as *u8); ko = eq_cat(key, ko, code)
90
91 let val: *u8 = sys_mmap(eq_len(code) + eq_len(file) + eq_len(detail) + eq_len(rung) + 64)
92 var vo: i64 = eq_cat(val, 0, code)
93 val[vo] = 9 as u8; vo = vo + 1; vo = eq_cat(val, vo, file)
94 val[vo] = 9 as u8; vo = vo + 1; vo = eq_cat(val, vo, detail)
95 val[vo] = 9 as u8; vo = vo + 1; vo = eq_cat(val, vo, "OPEN\x00" as *u8)
96 val[vo] = 9 as u8; vo = vo + 1; vo = eq_cat(val, vo, "eval\x00" as *u8)
97 val[vo] = 9 as u8; vo = vo + 1; vo = eq_cat(val, vo, rung)
98
99 let rc: i64 = eq_put(prefix, key, val)
100 if rc < 0 { return rc }
101
102 // maintain ws:ids (idempotent membership)
103 let pq: *i64 = sys_mmap(16) as *i64; let lq: *i64 = sys_mmap(16) as *i64
104 let idxbuf: *u8 = sys_mmap(K_MAGIC_65536); var io: i64 = 0
105 if ss_get(prefix, "ws:ids\x00" as *u8, pq, lq) == 1 {
106 let eb: *u8 = pq[0] as *u8; var e: i64 = 0
107 while e < lq[0] { idxbuf[io] = eb[e]; io = io + 1; e = e + 1 }
108 }
109 if eq_member(idxbuf, io, key) == 0 {
110 if io > 0 { idxbuf[io] = 9 as u8; io = io + 1 }
111 var t: i64 = 0; while key[t] != (0 as u8) { idxbuf[io] = key[t]; io = io + 1; t = t + 1 }
112 idxbuf[io] = 0 as u8
113 eq_put(prefix, "ws:ids\x00" as *u8, idxbuf)
114 }
115 return rc
116}
117
118// count queued gaps (TAB-count of ws:ids + 1, or 0 if empty).
119func nx_eval_queue_count() -> i64 {
120 let pq: *i64 = sys_mmap(16) as *i64; let lq: *i64 = sys_mmap(16) as *i64
121 if ss_get(EQ_PREFIX(), "ws:ids\x00" as *u8, pq, lq) != 1 { return 0 }
122 if lq[0] == 0 { return 0 }
123 let b: *u8 = pq[0] as *u8; var n: i64 = 1; var i: i64 = 0
124 while i < lq[0] { if b[i] == (9 as u8) { n = n + 1 } i = i + 1 }
125 return n
126}
127
128// enumerate: copy the Nth TAB-field of ws:ids (the Nth item's key) into out. returns key len, or 0 if out of range.
129// (makes the queue DRAINABLE -- a processor can walk items 0..count and act on each.)
130func nx_eval_queue_key(n: i64, out: *u8) -> i64 {
131 let pq: *i64 = sys_mmap(16) as *i64; let lq: *i64 = sys_mmap(16) as *i64
132 if ss_get(EQ_PREFIX(), "ws:ids\x00" as *u8, pq, lq) != 1 { return 0 }
133 let b: *u8 = pq[0] as *u8; let total: i64 = lq[0]
134 var field: i64 = 0; var i: i64 = 0; var o: i64 = 0
135 while i < total {
136 if b[i] == (9 as u8) { if field == n { out[o] = 0 as u8; return o } field = field + 1; o = 0 }
137 else { if field == n { out[o] = b[i]; o = o + 1 } }
138 i = i + 1
139 }
140 if field == n { out[o] = 0 as u8; return o }
141 return 0
142}
143
144// read an item's status field (field 3 of its TAB record: code\tfile\tdetail\tSTATUS\t...). copies into out.
145func nx_eval_queue_status(key: *u8, out: *u8) -> i64 {
146 let pq: *i64 = sys_mmap(16) as *i64; let lq: *i64 = sys_mmap(16) as *i64
147 if ss_get(EQ_PREFIX(), key, pq, lq) != 1 { out[0] = 0 as u8; return 0 }
148 let b: *u8 = pq[0] as *u8; let n: i64 = lq[0]
149 var field: i64 = 0; var i: i64 = 0; var o: i64 = 0
150 while i < n {
151 if b[i] == (9 as u8) { if field == 3 { out[o] = 0 as u8; return o } field = field + 1 }
152 else { if field == 3 { out[o] = b[i]; o = o + 1 } }
153 i = i + 1
154 }
155 out[o] = 0 as u8; return o
156}
157
158// CLOSE an item: rewrite its record with a new status (field 3) -> OPEN becomes DONE/DEAD-LETTER/IN-REVIEW.
159// This DRAINS the queue: an item that's been dispositioned is no longer OPEN, so drain-progress is real + durable.
160func nx_eval_queue_close(key: *u8, new_status: *u8) -> i64 {
161 let pq: *i64 = sys_mmap(16) as *i64; let lq: *i64 = sys_mmap(16) as *i64
162 if ss_get(EQ_PREFIX(), key, pq, lq) != 1 { return -1 }
163 let b: *u8 = pq[0] as *u8; let n: i64 = lq[0]
164 // rebuild: field0\tfield1\tfield2\t<new_status>\t<rest from field4 on>
165 let nv: *u8 = sys_mmap(K_MAGIC_2048); var o: i64 = 0
166 var field: i64 = 0; var i: i64 = 0
167 while i < n {
168 if b[i] == (9 as u8) {
169 if field == 2 {
170 nv[o] = 9 as u8; o = o + 1
171 var s: i64 = 0; while new_status[s] != (0 as u8) { nv[o] = new_status[s]; o = o + 1; s = s + 1 }
172 // skip the OLD field-3 status bytes until the next TAB
173 i = i + 1
174 while i < n { if b[i] == (9 as u8) { break } i = i + 1 }
175 field = 3
176 continue
177 }
178 nv[o] = 9 as u8; o = o + 1; field = field + 1
179 } else { nv[o] = b[i]; o = o + 1 }
180 i = i + 1
181 }
182 nv[o] = 0 as u8
183 return eq_put(EQ_PREFIX(), key, nv)
184}
185
186// ---- gate: emit gaps, read back, prove idempotency + isolation + liar-kill ----
187func main() -> i64 {
188 eq_puts("=== nx_eval_queue -- REPORT->LOG-NEXT bridge: evaluator gaps -> WMS work-signals (isolated queue) ===\n" as *u8)
189 let before: i64 = nx_eval_queue_count()
190 eq_puts(" queue before: " as *u8); eq_putn(before); eq_puts(" gaps\n" as *u8)
191
192 // emit two real gaps the evaluator found this session (skeletons = orphaned incomplete work to COMPLETE).
193 nx_eval_queue_emit("R5\x00" as *u8, "runtime/nx_ml_dsa_65.nx\x00" as *u8, "SKELETON_ERR_PENDING\x00" as *u8, "orphaned skeleton -> complete or designate _wasm canonical (never delete)\x00" as *u8)
194 nx_eval_queue_emit("R5\x00" as *u8, "runtime/nx_ml_kem_768.nx\x00" as *u8, "SKELETON_ERR_PENDING\x00" as *u8, "orphaned skeleton -> fill to FIPS-203 KAT or designate _wasm canonical\x00" as *u8)
195 let after1: i64 = nx_eval_queue_count()
196 eq_puts(" after emitting 2 gaps: " as *u8); eq_putn(after1); eq_puts(" gaps\n" as *u8)
197
198 // re-emit the SAME gap -> must NOT grow the queue (idempotent)
199 nx_eval_queue_emit("R5\x00" as *u8, "runtime/nx_ml_dsa_65.nx\x00" as *u8, "SKELETON_ERR_PENDING\x00" as *u8, "orphaned skeleton -> complete or designate _wasm canonical (never delete)\x00" as *u8)
200 let after2: i64 = nx_eval_queue_count()
201 eq_puts(" after re-emitting an identical gap: " as *u8); eq_putn(after2); eq_puts(" gaps (idempotent iff unchanged)\n" as *u8)
202
203 // read one back to prove the round-trip
204 let pq: *i64 = sys_mmap(16) as *i64; let lq: *i64 = sys_mmap(16) as *i64
205 let got: i64 = ss_get(EQ_PREFIX(), "gap:runtime/nx_ml_dsa_65.nx:SKELETON_ERR_PENDING\x00" as *u8, pq, lq)
206 var roundtrip: i64 = 0
207 if got == 1 { if lq[0] > 0 { roundtrip = 1 } }
208
209 // liar-kill: a never-emitted key must NOT be found
210 let ghost: i64 = ss_get(EQ_PREFIX(), "gap:runtime/nx_never_emitted_zzz9.nx:X\x00" as *u8, pq, lq)
211
212 eq_puts("\n ---- QUEUE VERDICT ----\n" as *u8)
213 var ok: i64 = 1
214 if after1 < before + 2 { ok = 0 } // two distinct gaps must have been added
215 if after2 != after1 { ok = 0 } // re-emit must NOT grow the queue
216 if roundtrip != 1 { ok = 0 } // the record must read back
217 if ghost == 1 { ok = 0 } // a ghost key must not exist
218 if ok == 1 {
219 eq_puts(" PASS: 2 gaps queued, re-emit idempotent, record round-trips, ghost absent -> REPORT->LOG-NEXT link PROVEN\n" as *u8)
220 eq_puts("NX-EVAL-QUEUE GREEN: detected gaps become queued work automatically (isolated store; conductor can pull -> DO). Loop QUEUE-gap CLOSED in composable form.\n" as *u8)
221 sys_exit(0); return 0
222 }
223 eq_puts(" before=" as *u8); eq_putn(before); eq_puts(" after1=" as *u8); eq_putn(after1); eq_puts(" after2=" as *u8); eq_putn(after2); eq_puts(" roundtrip=" as *u8); eq_putn(roundtrip); eq_puts(" ghost=" as *u8); eq_putn(ghost); eq_puts("\n" as *u8)
224 eq_puts("NX-EVAL-QUEUE RED\n" as *u8); sys_exit(1); return 1
225}