code wiki / _hdl_build / nx_express_lane.nx
nx_express_lane.nx source
↩ module page · 300 lines · 14451 B
1// nx_express_lane.nx -- X-XL-002 rung 1: the tutor<->team EXPRESS LANE consumer
2// (operator 2026-06-12: the team must not be a BLACK BOX -- S-class communication
3// everywhere including the command line; tutor tokens go to NOVEL work, not greps).
4// PLAIN LANGUAGE: the tutor (any model) appends verb rows to tutor_in.tsv; THIS
5// organ consumes them EXACTLY ONCE each beat, acts, ACKs every row to team_out.tsv,
6// then writes a SITREP block -- the one-read picture (quantitative lines verbatim
7// from the ledgers + queue counts). A tutor that still greps after this is
8// measured: every legitimate read is a channel row; archaeology shows.
9// verbs: MARKWIRE <id> <log::anchor::green> -> append ||MARK= to that row's gate
10// column (idempotent: a row already carrying ||MARK= is never re-marked);
11// ANSWER/ASK/VISIT-START/VISIT-END -> ACKed (consumption rungs follow:
12// X-XL-004 joins ANSWERs to briefs)
13// exactly-once: ACK key=<epoch>:<verb>:<id> in team_out is the cursor -- a row
14// whose key exists is never re-dispatched (re-run = zero re-work).
15// SITREP block: PULSE-VERDICT + EXAMINER-ARCS + AUTMETER + PAPERGEN-SUM lines
16// VERBATIM (ledger-derived, refuse-to-paraphrase) + queue status counts.
17// Durable: ACK + SITREP -> knowledge/channel/team_out.tsv. Exit 0 = lane duty done;
18// exit 1 = could not read its world. argv[1]=tutor_in argv[2]=team_out argv[3]=queue
19// (gates use scratch; live logs read at fixed paths only when argv absent).
20// license_tier: ORIGINAL
21import "nx_registry_lock.nx" // lib-only (nx_syscalls transitive); rt_lock/rt_unlock for the shared queue lock
22const K_MAGIC_1048576: i64 = 1048576
23const K_MAGIC_1048560: i64 = 1048560
24const K_MAGIC_262144: i64 = 262144
25const K_MAGIC_262128: i64 = 262128
26const K_MAGIC_16384: i64 = 16384
27func _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 _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 _fn(fd: i64, v: i64) -> i64 { let bb: *u8=sys_mmap(28); var m: i64=v; if m<0{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{bb[i]=t[k-1-i];i=i+1}; sys_write(fd,bb,k); return 0 }
30func el_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
31func el_cat(dst: *u8, off: i64, s: *u8) -> i64 { var i: i64 = 0; while s[i] != (0 as u8) { dst[off+i] = s[i]; i = i + 1 } return off + i }
32func el_read(path: *u8, buf: *u8, cap: i64) -> i64 {
33 let fd: i64 = sys_openat_rd(path)
34 if fd < 0 { return 0 }
35 var n: i64 = 0
36 var go: i64 = 1
37 while go == 1 { let base: i64 = buf as i64; let r: i64 = sys_read(fd, (base + n) as *u8, cap - n); if r <= 0 { go = 0 } else { n = n + r } if n >= cap { go = 0 } }
38 sys_close(fd)
39 return n
40}
41func el_slice_has(hay: *u8, a: i64, b: i64, pat: *u8) -> i64 {
42 let pl: i64 = el_len(pat)
43 if pl == 0 { return 0 }
44 var i: i64 = a
45 while i + pl <= b {
46 var k: i64 = 0
47 var hit: i64 = 1
48 while k < pl { if hay[i+k] != pat[k] { hit = 0; k = pl } else { k = k + 1 } }
49 if hit == 1 { return 1 }
50 i = i + 1
51 }
52 return 0
53}
54func el_last_line_with(hay: *u8, n: i64, pat: *u8, out: *i64) -> i64 {
55 var found: i64 = 0
56 var i: i64 = 0
57 while i < n {
58 var le: i64 = i
59 var s: i64 = 1
60 while s == 1 {
61 if le >= n { s = 0 }
62 else { if hay[le] == (10 as u8) { s = 0 } else { le = le + 1 } }
63 }
64 if el_slice_has(hay, i, le, pat) == 1 { out[0] = i; out[1] = le; found = 1 }
65 i = le + 1
66 }
67 return found
68}
69// copy the TAB-delimited token starting at pos (within [pos,le)) into pool; return end pos
70func el_tok(buf: *u8, pos: i64, le: i64, pool: *u8, poff: *i64) -> i64 {
71 var t: i64 = pos
72 var s: i64 = 1
73 while s == 1 {
74 if t >= le { s = 0 }
75 else { if buf[t] == (9 as u8) { s = 0 } else { pool[poff[0]] = buf[t]; poff[0] = poff[0] + 1; t = t + 1 } }
76 }
77 pool[poff[0]] = 0 as u8; poff[0] = poff[0] + 1
78 return t
79}
80// append ||MARK=<payload[pa,pb)> to the gate column (before the 8th TAB) of the row
81// whose line starts <id>TAB. 1=applied 2=already-marked 0=row-not-found/io
82func el_apply_mark(qp: *u8, ids: *u8, tin: *u8, pa: i64, pb: i64) -> i64 {
83 let qb: *u8 = sys_mmap(K_MAGIC_1048576)
84 let qn: i64 = el_read(qp, qb, K_MAGIC_1048560)
85 if qn <= 0 { return 0 }
86 let il: i64 = el_len(ids)
87 var i: i64 = 0
88 var at: i64 = 1
89 var ls: i64 = 0 - 1
90 var leF: i64 = 0 - 1
91 while i < qn {
92 if at == 1 {
93 if i + il < qn {
94 var k: i64 = 0
95 var hit: i64 = 1
96 while k < il { if qb[i+k] != ids[k] { hit = 0; k = il } else { k = k + 1 } }
97 if hit == 1 { if qb[i+il] == (9 as u8) {
98 ls = i
99 var q: i64 = i
100 var s2: i64 = 1
101 while s2 == 1 { if q >= qn { s2 = 0 } else { if qb[q] == (10 as u8) { s2 = 0 } else { q = q + 1 } } }
102 leF = q
103 i = qn
104 } }
105 }
106 }
107 if i < qn { if qb[i] == (10 as u8) { at = 1 } else { at = 0 } }
108 i = i + 1
109 }
110 if ls < 0 { return 0 }
111 if el_slice_has(qb, ls, leF, "||MARK=" as *u8) == 1 { return 2 }
112 // find the 8th TAB within the line (end of the gate column)
113 var tabs: i64 = 0
114 var p: i64 = ls
115 var ins: i64 = 0 - 1
116 while p < leF {
117 if qb[p] == (9 as u8) {
118 tabs = tabs + 1
119 if tabs == 8 { ins = p; p = leF }
120 }
121 p = p + 1
122 }
123 if ins < 0 { return 0 }
124 // tmp MUST be a SIBLING of the queue (same filesystem -- rename(2) across
125 // filesystems fails EXDEV; the gate caught exactly this lie on first run)
126 let tmp: *u8 = sys_mmap(512)
127 var to: i64 = el_cat(tmp, 0, qp)
128 to = el_cat(tmp, to, ".tmpmark" as *u8)
129 tmp[to] = 0 as u8
130 let fd: i64 = sys_openat_wr(tmp, 0x1a4)
131 if fd < 0 { return 0 }
132 sys_write(fd, qb, ins)
133 _fp(fd, "||MARK=" as *u8)
134 let base: i64 = tin as i64
135 sys_write(fd, (base + pa) as *u8, pb - pa)
136 let qbase: i64 = qb as i64
137 sys_write(fd, (qbase + ins) as *u8, qn - ins)
138 sys_close(fd)
139 if sys_renameat(tmp, qp) < 0 { return 0 }
140 return 1
141}
142func el_counts(qb: *u8, qn: i64, lfd: i64) -> i64 {
143 var todo: i64 = 0
144 var novel: i64 = 0
145 var done: i64 = 0
146 var i: i64 = 0
147 while i < qn {
148 var le: i64 = i
149 var s: i64 = 1
150 while s == 1 { if le >= qn { s = 0 } else { if qb[le] == (10 as u8) { s = 0 } else { le = le + 1 } } }
151 if el_slice_has(qb, i, le, "\tTODO\t" as *u8) == 1 { todo = todo + 1 }
152 if el_slice_has(qb, i, le, "\tNOVEL\t" as *u8) == 1 { novel = novel + 1 }
153 if el_slice_has(qb, i, le, "\tDONE\t" as *u8) == 1 { done = done + 1 }
154 i = le + 1
155 }
156 _fp(lfd, "SITREP-LINE queue todo=" as *u8); _fn(lfd, todo)
157 _fp(lfd, " novel=" as *u8); _fn(lfd, novel)
158 _fp(lfd, " done=" as *u8); _fn(lfd, done)
159 _fp(lfd, "\n" as *u8)
160 return 0
161}
162func el_sitline(lfd: i64, path: *u8, anchor: *u8, lb: *u8) -> i64 {
163 let ln: i64 = el_read(path, lb, K_MAGIC_1048560)
164 let span: *i64 = sys_mmap(16) as *i64
165 if ln > 0 {
166 if el_last_line_with(lb, ln, anchor, span) == 1 {
167 _fp(lfd, "SITREP-LINE " as *u8)
168 let base: i64 = lb as i64
169 sys_write(lfd, (base + span[0]) as *u8, span[1] - span[0])
170 _fp(lfd, "\n" as *u8)
171 return 1
172 }
173 }
174 _fp(lfd, "SITREP-LINE ABSENT anchor=" as *u8)
175 _fp(lfd, anchor)
176 _fp(lfd, "\n" as *u8)
177 return 0
178}
179func main(argc: i64, argv: *i64) -> i64 {
180 var tip: *u8 = "knowledge/channel/tutor_in.tsv" as *u8
181 var top: *u8 = "knowledge/channel/team_out.tsv" as *u8
182 var qp: *u8 = "knowledge/registry/assignment_queue.tsv" as *u8
183 var live: i64 = 1
184 if argc >= 2 { tip = argv[1] as *u8; live = 0 }
185 if argc >= 3 { top = argv[2] as *u8 }
186 if argc >= 4 { qp = argv[3] as *u8 }
187 _p("=== EXPRESS LANE: consume verbs exactly-once, ACK all, write the SITREP ===\n" as *u8)
188 let tin: *u8 = sys_mmap(K_MAGIC_262144)
189 let tn: i64 = el_read(tip, tin, K_MAGIC_262128)
190 if tn <= 0 { _p(" tutor_in MISSING -- fail loud\n" as *u8); sys_exit(1); return 1 }
191 let tout: *u8 = sys_mmap(K_MAGIC_262144)
192 let on: i64 = el_read(top, tout, K_MAGIC_262128)
193 // WMS torn-write fix: serialize the WHOLE team_out write session (the ACK lines AND the
194 // multi-LINE SITREP block) under the shared team_out advisory lock, so a concurrent writer
195 // (_express_lane_gate_authored's ELGATE row, or an overlapping lane run) can never interleave
196 // its records into ours. Per-line atomicity is insufficient here -- SITREP is many lines that
197 // must stay contiguous. Mirrors the queue-lock idiom below; degrade-gracefully on giveup (-1 ->
198 // proceed unlocked). Lock ordering top.lock -> qp.lock matches the inner MARKWIRE queue lock
199 // (consistent nesting, no deadlock).
200 let tolkp: *u8 = sys_mmap(512)
201 var tolko: i64 = el_cat(tolkp, 0, top); tolko = el_cat(tolkp, tolko, ".lock" as *u8); tolkp[tolko] = 0 as u8
202 let tolkfd: i64 = rt_lock(tolkp)
203 let lfd: i64 = sys_openat_append(top, 0x1a4)
204 if lfd < 0 { _p(" team_out open failed\n" as *u8); if tolkfd >= 0 { rt_unlock(tolkp, tolkfd) } sys_exit(1); return 1 }
205 let pool: *u8 = sys_mmap(K_MAGIC_16384)
206 let poffb: *i64 = sys_mmap(16) as *i64
207 poffb[0] = 0
208 let keyb: *u8 = sys_mmap(512)
209 var acks: i64 = 0
210 var marks: i64 = 0
211 var i: i64 = 0
212 while i < tn {
213 var le: i64 = i
214 var s: i64 = 1
215 while s == 1 { if le >= tn { s = 0 } else { if tin[le] == (10 as u8) { s = 0 } else { le = le + 1 } } }
216 if tin[i] != (35 as u8) {
217 if le > i + 8 {
218 let ep: i64 = (pool as i64) + poffb[0]
219 var t: i64 = el_tok(tin, i, le, pool, poffb)
220 let mdl: i64 = (pool as i64) + poffb[0]
221 t = el_tok(tin, t + 1, le, pool, poffb)
222 let vrb: i64 = (pool as i64) + poffb[0]
223 t = el_tok(tin, t + 1, le, pool, poffb)
224 let idt: i64 = (pool as i64) + poffb[0]
225 t = el_tok(tin, t + 1, le, pool, poffb)
226 let pa: i64 = t + 1
227 var o: i64 = el_cat(keyb, 0, "key=" as *u8)
228 o = el_cat(keyb, o, ep as *u8)
229 o = el_cat(keyb, o, ":" as *u8)
230 o = el_cat(keyb, o, vrb as *u8)
231 o = el_cat(keyb, o, ":" as *u8)
232 o = el_cat(keyb, o, idt as *u8)
233 keyb[o] = 0 as u8
234 if el_slice_has(tout, 0, on, keyb) == 0 {
235 var rc: i64 = 0
236 var note: *u8 = "received" as *u8
237 if el_slice_has(tin, i, le, "\tMARKWIRE\t" as *u8) == 1 {
238 // serialize the queue RMW (el_apply_mark reads + sibling-tmp-renames the
239 // queue) via the SHARED queue lock (qp + ".lock"); degrade-gracefully on
240 // giveup (-1 -> proceed unlocked; el_apply_mark stays atomic-rename either way).
241 let elkp: *u8 = sys_mmap(512)
242 var elko: i64 = el_cat(elkp, 0, qp); elko = el_cat(elkp, elko, ".lock" as *u8); elkp[elko] = 0 as u8
243 let elkfd: i64 = rt_lock(elkp)
244 let ar: i64 = el_apply_mark(qp, idt as *u8, tin, pa, le)
245 if elkfd >= 0 { rt_unlock(elkp, elkfd) }
246 if ar == 1 { rc = 0; note = "mark-applied" as *u8; marks = marks + 1 }
247 if ar == 2 { rc = 0; note = "already-marked" as *u8 }
248 if ar == 0 { rc = 1; note = "row-not-found" as *u8 }
249 }
250 _fp(lfd, "ACK " as *u8)
251 _fp(lfd, keyb)
252 _fp(lfd, " model=" as *u8)
253 _fp(lfd, mdl as *u8)
254 _fp(lfd, " rc=" as *u8); _fn(lfd, rc)
255 _fp(lfd, " note=" as *u8)
256 _fp(lfd, note)
257 _fp(lfd, "\n" as *u8)
258 acks = acks + 1
259 }
260 }
261 }
262 i = le + 1
263 }
264 // the SITREP block -- the one-read picture, lines VERBATIM from the ledgers
265 _fp(lfd, "SITREP epoch=" as *u8); _fn(lfd, sys_now_realtime_sec()); _fp(lfd, "\n" as *u8)
266 if live == 1 {
267 let lb: *u8 = sys_mmap(K_MAGIC_1048576)
268 el_sitline(lfd, "knowledge/status/team_pulse.log" as *u8, "PULSE-VERDICT" as *u8, lb)
269 el_sitline(lfd, "knowledge/status/examiner_arcs.log" as *u8, "EXAMINER-ARCS" as *u8, lb)
270 el_sitline(lfd, "knowledge/status/autonomy_meter.log" as *u8, "AUTMETER" as *u8, lb)
271 el_sitline(lfd, "knowledge/status/paper_gen.log" as *u8, "PAPERGEN-SUM" as *u8, lb)
272 el_sitline(lfd, "knowledge/status/tempo.log" as *u8, "TEMPO" as *u8, lb)
273 el_sitline(lfd, "knowledge/status/frontier.log" as *u8, "FRONTIER epoch" as *u8, lb)
274 el_sitline(lfd, "knowledge/status/surface_census.log" as *u8, "SURFACE-CENSUS" as *u8, lb)
275 el_sitline(lfd, "knowledge/status/tutor_necessity.log" as *u8, "TUTORNEED" as *u8, lb)
276 el_sitline(lfd, "knowledge/status/lang_maturity.log" as *u8, "LANGMAT epoch" as *u8, lb)
277 el_sitline(lfd, "knowledge/status/perf_panel.log" as *u8, "PERFPANEL epoch" as *u8, lb)
278 }
279 let qb2: *u8 = sys_mmap(K_MAGIC_1048576)
280 let qn2: i64 = el_read(qp, qb2, K_MAGIC_1048560)
281 if qn2 > 0 { el_counts(qb2, qn2, lfd) }
282 _fp(lfd, "SITREP-END acks=" as *u8); _fn(lfd, acks)
283 _fp(lfd, " marks_applied=" as *u8); _fn(lfd, marks)
284 _fp(lfd, "\n" as *u8)
285 sys_close(lfd)
286 if tolkfd >= 0 { rt_unlock(tolkp, tolkfd) } // release the team_out write session lock
287 _p(" LANE: acks=" as *u8)
288 let ob: *u8 = sys_mmap(28)
289 var m: i64 = acks
290 var k: i64 = 0
291 if m == 0 { ob[0] = 48 as u8; k = 1 }
292 while m > 0 { ob[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
293 let rb: *u8 = sys_mmap(28)
294 var z: i64 = 0
295 while z < k { rb[z] = ob[k-1-z]; z = z + 1 }
296 sys_write(1, rb, k)
297 _p(" -- SITREP appended to knowledge/channel/team_out.tsv (the one-read picture)\n" as *u8)
298 sys_exit(0)
299 return 0
300}