nx_supervised_dispatch_gate.nx source

1// nx_supervised_dispatch_gate.nx -- the REFEREE for WMS rung M5 (supervisor binding). 2// 3// module: nishi-core.autonomy.supervised_dispatch_gate 4// capability: GATE (prints a VERDICT computed from REAL checks; incl negative control + tamper) 5// 6// PROVES (the gate requirement, exactly): 7// T1 dispatch-binds -- M4 picks+leases a READY rung -> M5 forks a REAL supervised child 8// (dl_dispatch fd>=0 AND sd_spawn pid>0 AND job.state==RUNNING) 9// T2 limit-applied -- the job is resource-LIMITED (Job-Object-with-a-limit): a child 10// caps ITSELF via nx_prlimit(RLIMIT_AS) and the cap reads back == 11// SD_AS_LIMIT (proven by an in-organ probe child that exits with a 12// code that ENCODES "limit stuck", reaped here -- no fabrication) 13// T3 beats-while-alive-- a supervised job emits a heartbeat: hbm_scan at a fresh `now` 14// sees the spawned ws == HB_ALIVE (stalled==0) 15// T4 kill->stalled -- KILL the supervised job (nx_kill SIGKILL); advance injected `now` 16// past the threshold -> hbm_scan sees the missing heartbeat == 17// HB_STALLED (stalled==1). M1 detects the death. 18// T5 restart/reassign -- M5 restarts on the stall (the kernel/OS payoff): sd_supervise_tick 19// returns ACTION_RESTART(1), job.pid is NEW (!= killed pid), 20// restarts==1, and the fresh beat -> next hbm_scan == HB_ALIVE = 21// PROVEN BY A SECOND SUPERVISED RUN. 22// T6 NEG CONTROL -- a job that COMPLETES NORMALLY is NOT restarted (no thrash): spawn 23// a CLEAN payload, reap its exit(0) (state==EXITED_OK); then 24// sd_supervise_tick returns ACTION_NONE(0), restarts stays 0, no new 25// pid. A buggy M5 that restarts on "pid not alive" FAILS HERE. 26// T7 TAMPER -- restart storms are BOUNDED: drive 4 consecutive stalls (> 27// SD_MAXRESTART=3) -> tick returns ACTION_ESCALATE(3), no further 28// respawn = no infinite restart loop / car-alarm. 29// 30// VALIDITY (why the neg-control genuinely bites): T4 and T6 ride the SAME 31// sd_supervise_tick/hbm_scan path with OPPOSITE stimuli -- a killed job (no fresh beat -> 32// STALLED -> RESTART) vs a cleanly-exited job (state==EXITED_OK -> NONE). If M5 ignored the 33// heartbeat it FAILS T4; if it restarted on every dead pid it FAILS T6. The two lanes 34// cross-check the restart predicate exactly as the M1 gate cross-checks STALLED vs ALIVE. 35// 36// HERMETIC: own scratch heartbeat channel /tmp/sdg_<ms>.log + scratch queue /tmp/sdg_<ms>.tsv 37// + lease dir /tmp/sdg_<ms>/ (mirrors the M4 gate's g_scratch_paths). Time is INJECTED 38// (now/threshold = gate consts) so it is deterministic and re-runnable. 39// 40// WRITE DISCIPLINE: every evidence record = ONE buffer -> ONE locked fa_appendz to 41// knowledge/status/supervised_dispatch_gate.log AND a per-run UNIQUE tmp twin 42// (knowledge/status/supervised_dispatch_gate.<epoch>.<pid>.tmp) so concurrent gate runs 43// never contend (eat our own dogfood). Exit 0 GREEN / 1 RED. 44// 45// NO-FALSE-GREEN: green=1 ONLY if ALL of T1..T7 pass. If nx_prlimit is denied in this 46// sandbox (RLIMIT_AS EPERM) the gate reports RED + blocker, never a fabricated GREEN. 47// Sovereign: only nx_supervised_dispatch (+ M4/M1/framed_append transitively). No /bin. 48// license_tier: ORIGINAL 49import "nx_supervised_dispatch.nx" 50import "nx_framed_append.nx" 51 52const SDG_LOG: *u8 = "knowledge/status/supervised_dispatch_gate.log\x00" as *u8 53 54// injected time lattice (gate parameters, not baked into the organ): 55const SDG_BEAT0: i64 = 1000 // epoch of the FIRST (spawn-time) beat 56const SDG_THRESH: i64 = 60 // staleness threshold 57const SDG_NOW_HOT: i64 = 1010 // "now" while the job is fresh (age 10 < 60 -> ALIVE) 58const SDG_NOW_OLD: i64 = 2000 // "now" after the kill (age 1000 > 60 -> STALLED) 59const SDG_BEAT2: i64 = 2050 // epoch of the RESTART beat (fresh, > NOW_OLD - thresh) 60const SDG_NOW_AL2: i64 = 2060 // "now" after restart (age 10 < 60 -> ALIVE again) 61 62func g_p(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 63func g_n(v: i64) -> i64 { let bb: *u8=sys_mmap(28); var m: i64=v; if m<0{m=0-m; sys_write(1,"-\x00" as *u8,1)}; 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(1,bb,k); return 0 } 64 65// per-run UNIQUE tmp twin path (epoch+pid) so concurrent gate runs never contend. 66func g_unique_tmp(buf: *u8, epoch: i64, pid: i64) -> i64 { 67 var o: i64 = 0 68 o = fa_cat(buf, o, "knowledge/status/supervised_dispatch_gate.\x00" as *u8) 69 o = fa_catn(buf, o, epoch) 70 o = fa_cat(buf, o, ".\x00" as *u8) 71 o = fa_catn(buf, o, pid) 72 o = fa_cat(buf, o, ".tmp\x00" as *u8) 73 buf[o] = 0 as u8 74 return o 75} 76 77// one-buffer -> one locked fa_appendz evidence record, to BOTH the stable log and the 78// unique tmp twin: "SDGATE test=<name> result=PASS|FAIL detail=<n>" 79func g_log(twin: *u8, name: *u8, pass: i64, detail: i64) -> i64 { 80 let buf: *u8 = sys_mmap(256 + 16) 81 var o: i64 = 0 82 o = fa_cat(buf, o, "SDGATE test=\x00" as *u8) 83 o = fa_cat(buf, o, name) 84 o = fa_cat(buf, o, " result=\x00" as *u8) 85 if pass == 1 { o = fa_cat(buf, o, "PASS\x00" as *u8) } else { o = fa_cat(buf, o, "FAIL\x00" as *u8) } 86 o = fa_cat(buf, o, " detail=\x00" as *u8) 87 o = fa_catn(buf, o, detail) 88 buf[o] = 0 as u8 89 fa_appendz(SDG_LOG, buf, 256) 90 return fa_appendz(twin, buf, 256) 91} 92 93func g_assert(twin: *u8, name: *u8, pass: i64, detail: i64) -> i64 { 94 g_p(" " as *u8); g_p(name); g_p(": \x00" as *u8) 95 if pass == 1 { g_p("PASS\x00" as *u8) } else { g_p("FAIL\x00" as *u8) } 96 g_p(" (detail=\x00" as *u8); g_n(detail); g_p(")\n\x00" as *u8) 97 g_log(twin, name, pass, detail) 98 return pass 99} 100 101// build scratch paths: ldir = "/tmp/sdg_<ms>", qpath = ldir + ".tsv", hbpath = ldir + ".log" 102func g_scratch_paths(ms: i64, qpath: *u8, ldir: *u8, hbpath: *u8) -> i64 { 103 var o: i64 = 0 104 o = fa_cat(ldir, o, "/tmp/sdg_\x00" as *u8) 105 o = fa_catn(ldir, o, ms) 106 ldir[o] = 0 as u8 107 var q: i64 = 0 108 while ldir[q] != (0 as u8) { qpath[q] = ldir[q]; hbpath[q] = ldir[q]; q = q + 1 } 109 qpath[q] = 46 as u8; qpath[q+1] = 116 as u8; qpath[q+2] = 115 as u8; qpath[q+3] = 118 as u8; qpath[q+4] = 0 as u8 // ".tsv" 110 hbpath[q] = 46 as u8; hbpath[q+1] = 108 as u8; hbpath[q+2] = 111 as u8; hbpath[q+3] = 103 as u8; hbpath[q+4] = 0 as u8 // ".log" 111 return 0 112} 113 114// controlled scratch queue: ONE ready rung (the rung M5 dispatches+supervises) + a BLOCKED 115// rung (its dep is TODO) so the M4 pick is exercised honestly (never returns the blocked one). 116func g_write_fixture(path: *u8) -> i64 { 117 let fd: i64 = sys_openat_wr(path, 0x1a4) 118 if fd < 0 { return 0 - 1 } 119 let s: *u8 = "SUP-READY\tX\t9\tS\tBuilder\tTODO\t-\tgate\tsupervise-me\nSUP-BLOCKED\tX\t9\tS\tBuilder\tTODO\tSUP-DEP\tgate\tblocked\nSUP-DEP\tX\t1\tS\tBuilder\tTODO\t-\tgate\tunmet-dep\n\x00" as *u8 120 var n: i64 = 0 121 while s[n] != (0 as u8) { n = n + 1 } 122 sys_write(fd, s, n) 123 sys_close(fd) 124 return 0 125} 126 127// Reap a child (blocking wait) so the gate leaves no zombies after a kill/escalate. 128func g_reap(pid: i64) -> i64 { 129 let stp: *i64 = sys_mmap(16) as *i64 130 sys_wait4(pid, stp, 0) 131 return stp[0] 132} 133 134func main() -> i64 { 135 let pid_self: i64 = sd_getpid() 136 let now0: i64 = sys_now_realtime_sec() 137 g_p("=== WMS-M5 SUPERVISED-DISPATCH GATE: bind dispatch -> supervised job -> kill -> M1 stall -> restart; neg-ctrl clean-exit not restarted; tamper escalate ===\n\x00" as *u8) 138 139 let twin: *u8 = sys_mmap(160) 140 g_unique_tmp(twin, now0, pid_self) 141 142 let ms: i64 = sys_now_realtime_ms() 143 let qpath: *u8 = sys_mmap(160) 144 let ldir: *u8 = sys_mmap(160) 145 let hbpath:*u8 = sys_mmap(160) 146 g_scratch_paths(ms, qpath, ldir, hbpath) 147 sys_mkdir(ldir, 0x1ed) 148 if g_write_fixture(qpath) < 0 { g_p(" FIXTURE-WRITE-FAILED\n\x00" as *u8); sys_exit(1); return 1 } 149 150 let cx: *i64 = an_newcx() 151 let rows: i64 = an_load(qpath, cx) 152 if rows <= 0 { g_p(" FIXTURE-LOAD-FAILED\n\x00" as *u8); sys_exit(1); return 1 } 153 154 var pass: i64 = 0 155 let id_out: *u8 = sys_mmap(64) 156 let fd_out: *i64 = sys_mmap(16) as *i64 157 158 // ---- T2 FIRST (it must run before the prlimit-denied short-circuit): the limit child ---- 159 // Fork a probe child that caps ITSELF and exits with 42 IFF the cap read back == limit. 160 // This is the REAL "Job-Object-with-a-limit" proof, reaped here (no fabrication). If 161 // prlimit is denied (EPERM) the child exits with a DISTINCT code -> we honestly RED. 162 let lpid: i64 = sys_fork() 163 if lpid == 0 { 164 let enforced: i64 = sd_apply_limit(SD_AS_LIMIT) 165 if enforced == SD_AS_LIMIT { sys_exit(42) } // cap stuck exactly 166 if enforced < 0 { sys_exit(43) } // prlimit DENIED (EPERM) -> blocker 167 sys_exit(44) // cap stuck but to a different value 168 } 169 let lst: i64 = g_reap(lpid) 170 let lcode: i64 = (lst >> 8) & 0xFF 171 var t2: i64 = 0 172 if lcode == 42 { t2 = 1 } 173 if lcode == 43 { 174 // prlimit unavailable in this sandbox -> NO FALSE GREEN: report RED + precise blocker. 175 g_assert(twin, "T2-limit-applied\x00" as *u8, 0, lcode) 176 g_p("SUPERVISED-DISPATCH-GATE pass=0/7 verdict=RED blocker=prlimit-EPERM (RLIMIT_AS denied in sandbox)\n\x00" as *u8) 177 let bb: *u8 = sys_mmap(256+16) 178 var bo: i64 = 0 179 bo = fa_cat(bb, bo, "SUPERVISED-DISPATCH-GATE pass=0/7 verdict=RED blocker=prlimit-EPERM epoch=\x00" as *u8) 180 bo = fa_catn(bb, bo, sys_now_realtime_sec()) 181 bb[bo] = 0 as u8 182 fa_appendz(SDG_LOG, bb, 256); fa_appendz(twin, bb, 256) 183 sys_exit(1); return 1 184 } 185 pass = pass + g_assert(twin, "T2-limit-applied\x00" as *u8, t2, lcode) 186 187 // ---- T1: dispatch binds -> a REAL supervised child (CLEAN payload, will exit on its own) ---- 188 let job: *i64 = sd_job_new() 189 let pid1: i64 = sd_dispatch_and_supervise(cx, ldir, hbpath, job, 7, SD_PAY_CLEAN, 190 SDG_BEAT0, id_out, fd_out) 191 var t1: i64 = 0 192 if pid1 > 0 { if job[SD_J_STATE] == SD_RUNNING { if dl_streq(id_out, "SUP-READY\x00" as *u8) == 1 { t1 = 1 } } } 193 pass = pass + g_assert(twin, "T1-dispatch-binds\x00" as *u8, t1, pid1) 194 195 // ---- T3: beats-while-alive -- the supervised job's spawn-beat -> ws ALIVE at NOW_HOT ---- 196 let ids: *i64 = sys_mmap(8) as *i64 197 ids[0] = 7 198 let vd: *i64 = sys_mmap(8) as *i64 199 let outs: *i64 = sys_mmap(32) as *i64 200 hbm_scan(hbpath, SDG_NOW_HOT, SDG_THRESH, ids, 1, vd, outs) 201 var t3: i64 = 0 202 if vd[0] == HB_ALIVE { if outs[0] == 0 { t3 = 1 } } 203 pass = pass + g_assert(twin, "T3-beats-while-alive\x00" as *u8, t3, vd[0]) 204 205 // reap the T1 clean job so its exit is recorded (state -> EXITED_OK) for T6's no-thrash lane. 206 let st1: i64 = g_reap(pid1) 207 let c1: i64 = (st1 >> 8) & 0xFF 208 job[SD_J_STATE] = SD_EXITED_OK 209 job[SD_J_EXIT] = c1 210 211 // ====================================================================================== 212 // T4/T5: spawn a HANG job (never beats on its own; only the spawn-beat at BEAT0 exists), 213 // KILL it, advance injected `now` past threshold -> M1 STALLED -> M5 RESTART -> ALIVE. 214 // We use a SEPARATE ws lane (9) + a fresh job so the hang lane is independent of T1. 215 // ====================================================================================== 216 let hjob: *i64 = sd_job_new() 217 hjob[SD_J_WS] = 9 218 let kpid: i64 = sd_spawn_supervised(hbpath, hjob, SD_PAY_HANG, SDG_BEAT0) // spawn-beat @1000 219 // give the child a moment to actually be running before we kill it (bounded). 220 sys_sleep_ms(20) 221 222 // ---- T4: KILL the supervised job -> M1 sees the missing heartbeat ---- 223 let kr: i64 = nx_kill(kpid, SD_SIGKILL) 224 g_reap(kpid) // reap the killed child (no zombie) 225 let kids: *i64 = sys_mmap(8) as *i64 226 kids[0] = 9 227 let kvd: *i64 = sys_mmap(8) as *i64 228 let kouts: *i64 = sys_mmap(32) as *i64 229 // at NOW_OLD=2000 the only beat for ws9 is the spawn-beat @1000 -> age 1000 > 60 -> STALLED 230 hbm_scan(hbpath, SDG_NOW_OLD, SDG_THRESH, kids, 1, kvd, kouts) 231 var t4: i64 = 0 232 if kvd[0] == HB_STALLED { if kouts[0] == 1 { t4 = 1 } } 233 pass = pass + g_assert(twin, "T4-kill-stalled-detected\x00" as *u8, t4, kvd[0]) 234 235 // ---- T5: restart/reassign -- tick on the stall -> ACTION_RESTART, NEW pid, then ALIVE ---- 236 let old_pid: i64 = hjob[SD_J_PID] 237 // restart with a CLEAN payload so the second supervised run beats + completes (provable up). 238 let act5: i64 = sd_supervise_tick(hbpath, SDG_NOW_OLD, SDG_THRESH, hjob, SD_PAY_CLEAN, SDG_BEAT2) 239 let new_pid: i64 = hjob[SD_J_PID] 240 var t5: i64 = 0 241 if act5 == SD_ACTION_RESTART { 242 if new_pid != old_pid { 243 if new_pid > 0 { 244 if hjob[SD_J_RESTARTS] == 1 { 245 // the SECOND supervised run is ALIVE: its restart-beat @2050 clears the stall. 246 let r2ids: *i64 = sys_mmap(8) as *i64 247 r2ids[0] = 9 248 let r2vd: *i64 = sys_mmap(8) as *i64 249 let r2outs: *i64 = sys_mmap(32) as *i64 250 hbm_scan(hbpath, SDG_NOW_AL2, SDG_THRESH, r2ids, 1, r2vd, r2outs) 251 if r2vd[0] == HB_ALIVE { t5 = 1 } 252 } 253 } 254 } 255 } 256 g_reap(new_pid) // reap the (clean) restarted job 257 pass = pass + g_assert(twin, "T5-restart-second-supervised-run\x00" as *u8, t5, act5) 258 259 // ====================================================================================== 260 // T6 NEGATIVE CONTROL: a job that COMPLETED NORMALLY is NOT restarted (no thrash). 261 // The T1 job (state==EXITED_OK, reaped above) -> tick must return ACTION_NONE, no new pid, 262 // restarts stays 0. A buggy M5 that restarts on "pid not alive" would WRONGLY respawn here. 263 // ====================================================================================== 264 let before_pid: i64 = job[SD_J_PID] 265 let before_restarts: i64 = job[SD_J_RESTARTS] 266 let act6: i64 = sd_supervise_tick(hbpath, SDG_NOW_OLD, SDG_THRESH, job, SD_PAY_CLEAN, SDG_BEAT2) 267 var t6: i64 = 0 268 if act6 == SD_ACTION_NONE { if job[SD_J_PID] == before_pid { if job[SD_J_RESTARTS] == before_restarts { t6 = 1 } } } 269 pass = pass + g_assert(twin, "T6-NEGCTRL-clean-exit-not-restarted\x00" as *u8, t6, act6) 270 271 // ====================================================================================== 272 // T7 TAMPER: restart storms are BOUNDED. Drive consecutive STALLED readings past the 273 // budget and prove the policy stops respawning: restarts climb 1,2,3 then the 4th tick 274 // ESCALATES (no infinite restart loop / car-alarm). 275 // 276 // CORRECTNESS FIX (the M1 distinction the original T7 missed): a ws that has NEVER beaten 277 // reads HB_UNKNOWN, NOT HB_STALLED -- and sd_supervise_tick only RESTARTS on STALLED. So 278 // the job must have an OLD beat on record to read STALLED. We seed ONE old beat for ws11 279 // at SDG_BEAT0 (=1000); at the injected now=SDG_NOW_OLD (=2000) its age 1000 > thresh 60 -> 280 // STALLED. Each respawn uses SD_PAY_CLEAN at respawn_epoch=SDG_BEAT0: the restarted child 281 // exits IMMEDIATELY (no 100s HANG -> no gate hang) and its fresh beat is STAMPED at epoch 282 // 1000, which is STILL stale vs now=2000 -> the NEXT tick still reads STALLED. Thus the 283 // policy climbs 3 restarts then ESCALATEs on the 4th tick -- a clean, deterministic, 284 // hang-free TAMPER. We reap each spawned (clean, already-exited) child to avoid zombies. 285 // ====================================================================================== 286 let ejob: *i64 = sd_job_new() 287 ejob[SD_J_WS] = 11 288 let epid_self: i64 = sd_getpid() 289 hb_beat_at(hbpath, 11, SDG_BEAT0, 0, epid_self) // ONE old beat -> reads STALLED (not UNKNOWN) 290 var ti: i64 = 0 291 var esc_seen: i64 = 0 292 var restart_count: i64 = 0 293 while ti < 4 { 294 let ea: i64 = sd_supervise_tick(hbpath, SDG_NOW_OLD, SDG_THRESH, ejob, SD_PAY_CLEAN, SDG_BEAT0) 295 if ea == SD_ACTION_RESTART { restart_count = restart_count + 1; g_reap(ejob[SD_J_PID]) } 296 if ea == SD_ACTION_ESCALATE { esc_seen = 1 } 297 ti = ti + 1 298 } 299 var t7: i64 = 0 300 // exactly SD_MAXRESTART(3) restarts then ESCALATE on the 4th = bounded, no infinite loop. 301 if restart_count == SD_MAXRESTART { if esc_seen == 1 { if ejob[SD_J_STATE] == SD_ESCALATED { t7 = 1 } } } 302 pass = pass + g_assert(twin, "T7-TAMPER-escalate-budget-bounded\x00" as *u8, t7, restart_count) 303 304 // ---- VERDICT ---- 305 let total: i64 = 7 306 var ok: i64 = 0 307 if pass == total { ok = 1 } 308 g_p("SUPERVISED-DISPATCH-GATE pass=\x00" as *u8); g_n(pass); g_p("/\x00" as *u8); g_n(total); g_p(" verdict=\x00" as *u8) 309 if ok == 1 { g_p("GREEN\n\x00" as *u8) } else { g_p("RED\n\x00" as *u8) } 310 311 // release the M5 lease held since T1 (tidy hermetic state) 312 dl_release(ldir, "SUP-READY\x00" as *u8, fd_out[0]) 313 314 // final verdict record -- one buffer -> one locked fa_appendz (to BOTH stable log + twin) 315 let vb: *u8 = sys_mmap(256 + 16) 316 var vo: i64 = 0 317 vo = fa_cat(vb, vo, "SUPERVISED-DISPATCH-GATE pass=\x00" as *u8) 318 vo = fa_catn(vb, vo, pass) 319 vo = fa_cat(vb, vo, "/\x00" as *u8) 320 vo = fa_catn(vb, vo, total) 321 vo = fa_cat(vb, vo, " epoch=\x00" as *u8) 322 vo = fa_catn(vb, vo, sys_now_realtime_sec()) 323 vo = fa_cat(vb, vo, " verdict=\x00" as *u8) 324 if ok == 1 { vo = fa_cat(vb, vo, "GREEN\x00" as *u8) } else { vo = fa_cat(vb, vo, "RED\x00" as *u8) } 325 vb[vo] = 0 as u8 326 fa_appendz(SDG_LOG, vb, 256); fa_appendz(twin, vb, 256) 327 328 if ok == 1 { sys_exit(0); return 0 } 329 sys_exit(1) 330 return 1 331}