nx_pulse_worker.nx source
↩ module page · 327 lines · 12768 B
1// nx_pulse_worker.nx -- LANE WORKER: NAS-owned schedule -> laptop execution (operator 2026-07-16: the
2// autograde pulse "is supposed to be nishi ecosystem native using the nas" -- the Windows scheduled task
3// built earlier the same day was REJECTED and deleted; this organ replaces it).
4//
5// DESIGN (every piece an ecosystem piece; zero Windows constructs; zero inbound ports):
6// * THE NAS OWNS THE SCHEDULE AS DATA: https://nishifamily.com/evidence/autograde/pulse_policy.txt
7// (edited any time via /api/upload + /api/promote_content). Keys: enabled=0|1, mode=interval|daily,
8// interval_min=N, daily_utc_hour=H.
9// * THIS worker polls that policy OUTBOUND over sovereign TLS (forks _offc/nx_mgmt_client.elf GET --
10// the proven WSL->edge lane) and fires only when the NAS policy says due. The laptop is the executor
11// because the grader organs + ledgers live on this box's build lane; the schedule truth is the NAS's.
12// * A FIRE = fork the proven payload bench/autograde_pulse.sh (autograde gate -> dashboard ship -> the
13// script appends its own PULSE start/done rows). Worker appends a PWRK audit row per fire.
14// * FAIL-CLOSED: policy unreachable / status!=200 / enabled!=1 / unparsable => NO fire.
15// * RATE-LIMITED BY CONSTRUCTION: never fires within 1800s of the last PULSE start, whatever the
16// policy says -- a crashed ship or an operator typo (interval_min=0) cannot rapid-fire the loop.
17// * UPGRADE PATH: when the swarm worker-agent lane lands, this worker pulls hub-queued jobs instead of
18// a polled policy file -- same shape, richer plumbing (job kinds, permissions, QUEUE-when-busy).
19//
20// usage: nx_pulse_worker [poll_seconds] [oneshot] -- poll default 120; any 3rd arg = one cycle then exit.
21// MUST be launched from the nxc2 checkout root (relative _offc/ + bench/ paths).
22// HANG-PROOF (debt eaten 2026-07-16): both child forks ride nx_guarded_run (WNOHANG poll + wall-clock
23// deadline + SIGKILL + reap, gate-proven) -- a wedged TLS fetch or stuck payload can no longer freeze
24// the daemon; it is killed at the deadline and the cycle fails CLOSED (no fire on unknown policy).
25// Child stdout rides a pipe drained AFTER exit -- safe because both outputs are far below the 64KB
26// pipe buffer (policy ~1KB, pulse log ~4KB); a runaway writer blocks and the deadline reaps it.
27// expect_exit: 0 license_tier: ORIGINAL
28import "nx_guarded_run.nx"
29const K_MAGIC_999999: i64 = 999999
30const K_MAGIC_20000: i64 = 20000
31const K_MAGIC_65535: i64 = 65535
32const K_MAGIC_16000: i64 = 16000
33const K_MAGIC_1800: i64 = 1800
34const K_MAGIC_86400: i64 = 86400
35const K_MAGIC_3600: i64 = 3600
36const K_MAGIC_300000: i64 = 300000
37const K_MAGIC_262143: i64 = 262143
38const K_MAGIC_65536: i64 = 65536
39const K_MAGIC_16384: i64 = 16384
40const K_MAGIC_262144: i64 = 262144
41
42static g_ts: *i64
43static g_pol: *u8
44static g_led: *u8
45static g_out: *u8
46static g_av: *i64
47static g_olen: *i64
48static g_row: *u8
49static g_res: *i64
50
51func w(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
52func wn(v: i64) -> i64 {
53 var m: i64 = v
54 if m < 0 { w("-" as *u8); m = 0 - m }
55 let t: *u8 = sys_mmap(24)
56 var k: i64 = 0
57 if m == 0 { t[0] = 48 as u8; k = 1 }
58 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
59 let o: *u8 = sys_mmap(24)
60 var i: i64 = 0
61 while i < k { o[i] = t[k - 1 - i]; i = i + 1 }
62 sys_write(1, o, k)
63 sys_munmap(t, 24)
64 sys_munmap(o, 24)
65 return 0
66}
67func slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
68
69func find_last(hay: *u8, hn: i64, needle: *u8) -> i64 {
70 let m: i64 = slen(needle)
71 if m == 0 { return 0 - 1 }
72 var last: i64 = 0 - 1
73 var i: i64 = 0
74 while i + m <= hn {
75 var j: i64 = 0
76 var ok: i64 = 1
77 while j < m { if hay[i+j] != needle[j] { ok = 0; j = m } else { j = j + 1 } }
78 if ok == 1 { last = i }
79 i = i + 1
80 }
81 return last
82}
83
84// parse an integer at-or-just-after p in hay[0,hn): skips up to 12 non-digit chars, then reads [-]digits.
85// returns -999999 (sentinel) if no number found. Callers pass p = marker_start + slen(marker).
86func pint_at(hay: *u8, hn: i64, p: i64) -> i64 {
87 var i: i64 = p
88 var guard: i64 = 0
89 var found: i64 = 0
90 while found == 0 {
91 if i >= hn { return 0 - K_MAGIC_999999 }
92 if guard >= 12 { return 0 - K_MAGIC_999999 }
93 let c: i64 = hay[i] as i64
94 if c == 45 { found = 1 }
95 else { if c >= 48 { if c <= 57 { found = 1 } else { i = i + 1; guard = guard + 1 } } else { i = i + 1; guard = guard + 1 } }
96 }
97 var neg: i64 = 0
98 if (hay[i] as i64) == 45 { neg = 1; i = i + 1 }
99 var v: i64 = 0
100 var go: i64 = 1
101 while go == 1 {
102 if i >= hn { go = 0 }
103 else {
104 let c2: i64 = hay[i] as i64
105 if c2 >= 48 { if c2 <= 57 { v = v * 10 + (c2 - 48); i = i + 1 } else { go = 0 } } else { go = 0 }
106 }
107 }
108 if neg == 1 { v = 0 - v }
109 return v
110}
111
112func pw_atoi(s: *u8) -> i64 {
113 var v: i64 = 0
114 var i: i64 = 0
115 while s[i] != (0 as u8) {
116 let c: i64 = s[i] as i64
117 if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } }
118 i = i + 1
119 }
120 return v
121}
122
123func bcat(b: *u8, off: i64, s: *u8) -> i64 {
124 var i: i64 = 0
125 var o: i64 = off
126 while s[i] != (0 as u8) { b[o] = s[i]; o = o + 1; i = i + 1 }
127 return o
128}
129func bcatn(b: *u8, off: i64, v: i64) -> i64 {
130 var m: i64 = v
131 var o: i64 = off
132 if m < 0 { b[o] = 45 as u8; o = o + 1; m = 0 - m }
133 let t: *u8 = sys_mmap(24)
134 var k: i64 = 0
135 if m == 0 { t[0] = 48 as u8; k = 1 }
136 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
137 var i: i64 = 0
138 while i < k { b[o] = t[k - 1 - i]; o = o + 1; i = i + 1 }
139 sys_munmap(t, 24)
140 return o
141}
142
143func pw_now() -> i64 {
144 sys_clock_gettime_real(g_ts)
145 return g_ts[0]
146}
147
148// guarded fork+capture: run path/argv under a hard deadline, drain child stdout from a pipe after
149// exit. Returns the child's exit code (124 = deadline-killed); bytes captured via olenbox[0].
150func pw_run_guarded(path: *u8, argv: *i64, deadline_ms: i64, out: *u8, cap: i64, olenbox: *i64) -> i64 {
151 olenbox[0] = 0
152 let fds: *i64 = sys_mmap(16) as *i64
153 if sys_pipe2(fds, 0) != 0 { return 0 - 2 }
154 let packed: i64 = fds[0]
155 let rfd: i64 = packed & 0xFFFFFFFF
156 let wfd: i64 = (packed >> 32) & 0xFFFFFFFF
157 let envp: *i64 = sys_mmap(16) as *i64
158 envp[0] = ("PATH=/usr/bin:/bin" as *u8) as i64
159 envp[1] = 0
160 let rc: i64 = nx_guarded_run(path, argv, envp, deadline_ms, wfd, wfd)
161 sys_close(wfd)
162 var total: i64 = 0
163 var go: i64 = 1
164 while go == 1 {
165 if total >= cap { go = 0 }
166 else {
167 let r: i64 = sys_read(rfd, ((out as i64) + total) as *u8, cap - total)
168 if r <= 0 { go = 0 } else { total = total + r }
169 }
170 }
171 sys_close(rfd)
172 olenbox[0] = total
173 return rc
174}
175
176// fetch the NAS policy over sovereign TLS. Fills res[0]=enabled res[1]=mode(0 interval,1 daily)
177// res[2]=interval_min res[3]=daily_utc_hour. Returns 1 parsed-ok, 0 fail (=> fail-closed no-fire).
178func pw_fetch_policy(res: *i64) -> i64 {
179 g_av[0] = ("_offc/nx_mgmt_client.elf" as *u8) as i64
180 g_av[1] = ("https://nishifamily.com" as *u8) as i64
181 g_av[2] = ("call" as *u8) as i64
182 g_av[3] = ("GET" as *u8) as i64
183 g_av[4] = ("/evidence/autograde/pulse_policy.txt" as *u8) as i64
184 g_av[5] = ("bench/dummy_tok" as *u8) as i64
185 g_av[6] = 0
186 let rc: i64 = pw_run_guarded("_offc/nx_mgmt_client.elf" as *u8, g_av, K_MAGIC_20000, g_pol, K_MAGIC_65535, g_olen)
187 if rc == 124 { return 0 }
188 let n: i64 = g_olen[0]
189 if n <= 0 { return 0 }
190 if find_last(g_pol, n, "status=200" as *u8) < 0 { return 0 }
191 let ken: *u8 = "enabled=" as *u8
192 let pe: i64 = find_last(g_pol, n, ken)
193 if pe < 0 { return 0 }
194 res[0] = pint_at(g_pol, n, pe + slen(ken))
195 res[1] = 0
196 if find_last(g_pol, n, "mode=daily" as *u8) >= 0 { res[1] = 1 }
197 let kim: *u8 = "interval_min=" as *u8
198 let pi: i64 = find_last(g_pol, n, kim)
199 if pi >= 0 { res[2] = pint_at(g_pol, n, pi + slen(kim)) } else { res[2] = 0 - K_MAGIC_999999 }
200 let kdh: *u8 = "daily_utc_hour=" as *u8
201 let ph: i64 = find_last(g_pol, n, kdh)
202 if ph >= 0 { res[3] = pint_at(g_pol, n, ph + slen(kdh)) } else { res[3] = 0 - K_MAGIC_999999 }
203 return 1
204}
205
206// read the tail of the pulse ledger into g_led; returns bytes read (0 if absent).
207func pw_led_read() -> i64 {
208 let fd: i64 = sys_openat_rd("/home/elderwesto/nx_stage/autograde_pulse.log" as *u8)
209 if fd < 0 { return 0 }
210 let sz: i64 = sys_lseek(fd, 0, 2)
211 var off: i64 = 0
212 if sz > K_MAGIC_16000 { off = sz - K_MAGIC_16000 }
213 sys_lseek(fd, off, 0)
214 let n: i64 = sys_read(fd, g_led, K_MAGIC_16000)
215 sys_close(fd)
216 if n < 0 { return 0 }
217 return n
218}
219
220// last timestamp after `marker` in the ledger tail; 0 if none.
221func pw_last(n: i64, marker: *u8) -> i64 {
222 let p: i64 = find_last(g_led, n, marker)
223 if p < 0 { return 0 }
224 let v: i64 = pint_at(g_led, n, p + slen(marker))
225 if v == (0 - K_MAGIC_999999) { return 0 }
226 return v
227}
228
229// the due decision, PURE (gate-testable): policy res + now + last done/start -> 1 fire / 0 hold.
230func pw_due(res: *i64, now: i64, lastdone: i64, laststart: i64) -> i64 {
231 if res[0] != 1 { return 0 }
232 if now < laststart + K_MAGIC_1800 { return 0 }
233 if res[1] == 0 {
234 if res[2] == (0 - K_MAGIC_999999) { return 0 }
235 if res[2] <= 0 { return 0 }
236 if lastdone == 0 { return 1 }
237 if now >= lastdone + res[2] * 60 { return 1 }
238 return 0
239 }
240 if res[3] == (0 - K_MAGIC_999999) { return 0 }
241 if res[3] < 0 { return 0 }
242 if res[3] > 23 { return 0 }
243 let dnow: i64 = now / K_MAGIC_86400
244 let dlast: i64 = lastdone / K_MAGIC_86400
245 if dnow > dlast { if ((now % K_MAGIC_86400) / K_MAGIC_3600) >= res[3] { return 1 } }
246 return 0
247}
248
249// fire the proven payload under a 300s deadline; returns its exit code (124 = deadline-killed).
250func pw_fire() -> i64 {
251 g_av[0] = ("/bin/bash" as *u8) as i64
252 g_av[1] = ("bench/autograde_pulse.sh" as *u8) as i64
253 g_av[2] = 0
254 return pw_run_guarded("/bin/bash" as *u8, g_av, K_MAGIC_300000, g_out, K_MAGIC_262143, g_olen)
255}
256
257// append the worker's audit row beside the payload's own PULSE rows.
258func pw_append(now: i64, rc: i64) -> i64 {
259 var o: i64 = 0
260 o = bcat(g_row, o, "PWRK fired ts=" as *u8)
261 o = bcatn(g_row, o, now)
262 o = bcat(g_row, o, " rc=" as *u8)
263 o = bcatn(g_row, o, rc)
264 g_row[o] = 10 as u8
265 o = o + 1
266 let fd: i64 = sys_openat_append("/home/elderwesto/nx_stage/autograde_pulse.log" as *u8, 0x1a4)
267 if fd < 0 { return 0 }
268 sys_write(fd, g_row, o)
269 sys_close(fd)
270 return o
271}
272
273func main(argc: i64, argv: *i64) -> i64 {
274 g_ts = sys_mmap(16) as *i64
275 g_pol = sys_mmap(K_MAGIC_65536)
276 g_led = sys_mmap(K_MAGIC_16384)
277 g_out = sys_mmap(K_MAGIC_262144)
278 g_av = sys_mmap(64) as *i64
279 g_olen = sys_mmap(8) as *i64
280 g_row = sys_mmap(256)
281 g_res = sys_mmap(32) as *i64
282 var poll: i64 = 120
283 var oneshot: i64 = 0
284 if argc >= 2 { poll = pw_atoi(argv[1] as *u8) }
285 if poll <= 0 { poll = 120 }
286 if argc >= 3 { oneshot = 1 }
287 w("NX-PULSE-WORKER: NAS-owned schedule (/evidence/autograde/pulse_policy.txt) -> laptop execution\n" as *u8)
288 w(" poll=" as *u8)
289 wn(poll)
290 w("s oneshot=" as *u8)
291 wn(oneshot)
292 w(" (fail-closed: no policy => no fire; floor 1800s between fires)\n" as *u8)
293 var go: i64 = 1
294 while go == 1 {
295 let now: i64 = pw_now()
296 w("PW cycle now=" as *u8)
297 wn(now)
298 let ok: i64 = pw_fetch_policy(g_res)
299 if ok == 0 { w(" policy-fetch-FAIL (fail-closed: no fire)\n" as *u8) }
300 else {
301 let n: i64 = pw_led_read()
302 let ld: i64 = pw_last(n, "PULSE done ts=" as *u8)
303 let lst: i64 = pw_last(n, "PULSE start ts=" as *u8)
304 let d: i64 = pw_due(g_res, now, ld, lst)
305 w(" enabled=" as *u8)
306 wn(g_res[0])
307 w(" mode=" as *u8)
308 if g_res[1] == 1 { w("daily@" as *u8); wn(g_res[3]) } else { w("interval/" as *u8); wn(g_res[2]) }
309 w(" lastdone=" as *u8)
310 wn(ld)
311 w(" due=" as *u8)
312 wn(d)
313 w("\n" as *u8)
314 if d == 1 {
315 w("PW FIRING (the NAS policy says due)\n" as *u8)
316 let rc: i64 = pw_fire()
317 pw_append(now, rc)
318 w("PW fire done rc=" as *u8)
319 wn(rc)
320 w("\n" as *u8)
321 }
322 }
323 if oneshot == 1 { go = 0 }
324 else { sys_sleep_ms(poll * 1000) }
325 }
326 return 0
327}