nx_supervised_dispatch.nx source

1// nx_supervised_dispatch.nx -- WMS rung M5: OS / KERNEL SUPERVISOR BINDING. 2// 3// module: nishi-core.autonomy.supervised_dispatch 4// capability: BIND_DISPATCH_TO_SUPERVISED_JOB (a dispatched rung becomes a real, 5// resource-LIMITED, reaped, RESTART-on-death supervised job) 6// 7// WHAT THIS UNIFIES (the M5 payoff): M4 (nx_dispatch_lease) LEASES the next ready rung 8// but never RUNS it; M1 (nx_heartbeat_monitor) detects a stalled stream but never ACTS; 9// the supervisor policy (nx_supervisor) decides RESTART/ESCALATE but never forks. M5 10// is the missing seam: it FORKS the leased rung as a governed child (with a kernel 11// resource limit = prlimit64 RLIMIT_AS, the Linux analog of a Job-Object memory limit), 12// beats its heartbeat, and -- when M1 sees a missing heartbeat (a crashed job) -- it 13// RESTARTS/reassigns (a second supervised run). Conductor loop + kernel scheduler + 14// workstream manager collapse into ONE control plane. 15// 16// REUSE / lineage (compose, do not reinvent -- check-registries-before-building): 17// - dl_dispatch / dl_try_lease / dl_release <- nx_dispatch_lease.nx (M4): lease the rung 18// - an_newcx / an_load / an_find / an_id_at <- nx_assign_core (surfaced via M4) 19// - hb_beat_at / hbm_scan / HB_STALLED/HB_ALIVE <- nx_heartbeat_monitor.nx (M1) 20// - fa_cat / fa_catn / fa_appendz <- nx_framed_append.nx (R0b): the locked 21// single-write status discipline (every record = ONE buffer -> ONE atomic write) 22// - the fork -> child-setup -> sys_wait4 status-decode model is the nx_container.nx idiom 23// (lo = status & 0xFF; exited iff lo==0; code = (status>>8)&0xFF) -- reimplemented HERE 24// (NOT imported) because nx_container.nx imports nx_syscalls_x86_64.nx and that would 25// surface a SECOND syscall table against M1/M4's nx_syscalls.nx = the double-import 26// rc=6 trap. M5 keeps a SINGLE syscall surface (nx_syscalls.nx via all three imports). 27// 28// SOVEREIGN PAYLOAD (no /bin dependency): the supervised child does NOT execve a foreign 29// ELF. After fork it caps ITSELF (nx_prlimit) then runs an IN-ORGAN payload selector -- 30// the same organ post-fork executing a payload function (the nx_container fork model minus 31// execve). payload=0 clean work+exit(0); payload=1 "crash" exit(70); payload=2 hang 32// (finite busy then exit WITHOUT beating = the never-beats stall the gate KILLs/detects). 33// 34// SUPERVISOR POLICY (inlined, ~ nx_supervisor's 4-state restart-budget machine): a STALLED 35// job under budget -> RESTART (respawn + restarts++ + fresh beat); over budget -> ESCALATE 36// (bounded, no thrash storm); an ALIVE / clean-exited job -> NONE (the no-thrash neg-ctrl). 37// 38// WRITE DISCIPLINE (the torn-line bug we must NOT reintroduce): every status record is 39// assembled into ONE buffer and emitted with a SINGLE locked fa_appendz (the cn_emit_* 40// discipline) -- never a sequence of sys_write() calls. Additive: new file only; one tiny 41// additive nx_prlimit wrapper was added to nx_syscalls.nx (no existing function touched). 42// license_tier: ORIGINAL 43import "nx_dispatch_lease.nx" // dl_dispatch / dl_try_lease / dl_release (+ an_*, nx_syscalls, framed_append) 44import "nx_heartbeat_monitor.nx" // hb_beat_at / hbm_scan / HB_ALIVE / HB_STALLED (+ nx_ws_ledger) 45import "nx_framed_append.nx" // fa_cat / fa_catn / fa_appendz (status discipline) 46const SD_MAGIC_2277: i64 = 2277 47const SD_MAGIC_100000: i64 = 100000 48 49const SD_REC_CAP: i64 = 256 // bounded status record size (no magic number) 50const SD_SIGKILL: i64 = 9 // the crash signal for the controlled-kill test 51const SD_MAXRESTART: i64 = 3 // restart budget; over -> ESCALATE (no thrash storm) 52const SD_AS_LIMIT: i64 = 0x10000000 // 256 MiB address-space cap = the resource "limit" 53 54// supervisor actions (mirror nx_supervisor's NX_SUP_ACTION_* without importing it) 55const SD_ACTION_NONE: i64 = 0 // alive / clean-exited -> do nothing (no-thrash path) 56const SD_ACTION_RESTART: i64 = 1 // stalled + under budget -> respawn 57const SD_ACTION_ESCALATE: i64 = 3 // stalled + budget exhausted -> escalate (bounded) 58 59// job state lanes 60const SD_RUNNING: i64 = 0 61const SD_EXITED_OK: i64 = 1 62const SD_CRASHED: i64 = 2 63const SD_ESCALATED: i64 = 3 64 65// payload selectors (the in-organ supervised work; no /bin, no execve) 66const SD_PAY_CLEAN: i64 = 0 // compute + exit(0): a job that finishes normally 67const SD_PAY_CRASH: i64 = 1 // exit(70): a job that dies non-zero 68const SD_PAY_HANG: i64 = 2 // finite busy, exit WITHOUT beating (the stall payload) 69 70// SdJob descriptor as a flat i64 block (8 slots, no heap churn). Layout: 71// j[0]=ws_id j[1]=pid j[2]=restarts j[3]=state j[4]=limit_as j[5]=last_exit/sig 72const SD_J_WS: i64 = 0 73const SD_J_PID: i64 = 1 74const SD_J_RESTARTS: i64 = 2 75const SD_J_STATE: i64 = 3 76const SD_J_LIMIT: i64 = 4 77const SD_J_EXIT: i64 = 5 78 79func sd_job_new() -> *i64 { 80 let j: *i64 = sys_mmap(8 * 8) as *i64 81 j[SD_J_WS] = 0; j[SD_J_PID] = 0; j[SD_J_RESTARTS] = 0 82 j[SD_J_STATE] = SD_RUNNING; j[SD_J_LIMIT] = SD_AS_LIMIT; j[SD_J_EXIT] = 0 83 return j 84} 85 86func sd_getpid() -> i64 { return __syscall(172, 0, 0, 0, 0, 0, 0) } // rv64 getpid=172; raw x86 39 is an RV64 KEY translated to ioctl(16) -> -ENOTTY (debt idx SD_MAGIC_2277) 87 88// Apply a kernel resource limit (RLIMIT_AS) to SELF (pid=0). Called in the forked child 89// BEFORE the payload runs -> the supervised job is genuinely resource-LIMITED (the Job- 90// Object-with-a-limit). Reads the value back (old_limit) and returns the ENFORCED 91// rlim_cur so a caller/child can prove the cap actually stuck (T2). Returns -errno on 92// failure (e.g. EPERM in a locked-down sandbox -> the gate honestly reports RED, no fake). 93func sd_apply_limit(limit_as: i64) -> i64 { 94 let nl: *u8 = sys_mmap(16) // struct rlimit64 { cur, max } 95 let nlp: *i64 = nl as *i64 96 nlp[0] = limit_as // rlim_cur 97 nlp[1] = limit_as // rlim_max (soft==hard; lowering is unprivileged-OK) 98 // SET (old_limit=0: we do NOT want the previous value -- prlimit writes the OLD limit into 99 // old_limit, and the old RLIMIT_AS is RLIM_INFINITY=-1, which would falsely look like a 100 // failure). rc<0 => the cap was DENIED (e.g. EPERM in a locked sandbox) -> propagate. 101 let rc: i64 = nx_prlimit(0, RLIMIT_AS, nl, 0 as *u8) 102 if rc < 0 { return rc } 103 // READ BACK fresh (new=0, old=&cur): returns the NOW-ENFORCED rlim_cur -> the proof the 104 // cap actually stuck (the gate's T2 child asserts this == SD_AS_LIMIT). 105 let cur: *u8 = sys_mmap(16) 106 let rcr: i64 = nx_prlimit(0, RLIMIT_AS, 0 as *u8, cur) 107 if rcr < 0 { return rcr } 108 let curp: *i64 = cur as *i64 109 return curp[0] // the enforced rlim_cur (== limit_as on success) 110} 111 112// The sovereign supervised PAYLOAD (runs ONLY in the forked child; never returns). 113// beats: if 1 AND payload != HANG, the child emits ONE heartbeat for ws (proves a live 114// supervised job beats). The HANG payload never beats by design (the stall). 115func sd_child_payload(hbpath: *u8, ws: i64, payload: i64, limit_as: i64, beats: i64, 116 beat_epoch: i64) -> i64 { 117 let enforced: i64 = sd_apply_limit(limit_as) // cap SELF first (the resource limit) 118 if payload == SD_PAY_HANG { 119 // hang: a finite busy spin (sovereign, bounded -- the gate KILLs it before this 120 // ends, but it self-terminates if the kill is ever missed so no zombie wedge). 121 sys_sleep_ms(SD_MAGIC_100000) // ~100s; the gate's nx_kill arrives first 122 sys_exit(80) // (only reached if never killed) 123 return 80 124 } 125 if beats == 1 { 126 let pid: i64 = sd_getpid() 127 hb_beat_at(hbpath, ws, beat_epoch, 0, pid) // ONE locked beat = "this job is alive" 128 } 129 if payload == SD_PAY_CRASH { 130 sys_exit(70) // a job that dies non-zero 131 return 70 132 } 133 // SD_PAY_CLEAN: a small deterministic compute, then a normal exit(0). 134 var acc: i64 = 0 135 var i: i64 = 0 136 while i < 1000 { acc = acc + i; i = i + 1 } 137 if enforced < 0 { sys_exit(91) } // limit failed -> distinct non-zero 138 sys_exit(0) // finished normally (no restart wanted) 139 return 0 140} 141 142// FORK a supervised job: child caps itself + runs payload(ws); parent stamps pid + 143// RUNNING and returns the child pid (>0). The parent ALSO beats an initial heartbeat 144// for ws (beat_epoch) so a freshly-spawned job is immediately ALIVE to the M1 scan even 145// before the child's own beat lands (race-free liveness). Returns child pid, or -errno. 146func sd_spawn_supervised(hbpath: *u8, job: *i64, payload: i64, beat_epoch: i64) -> i64 { 147 let pid: i64 = sys_fork() 148 if pid < 0 { return pid } // fork failed -> -errno 149 if pid == 0 { 150 // CHILD: a clean job/crash beats; a hang never beats (gate detects the stall). 151 var beats: i64 = 1 152 if payload == SD_PAY_HANG { beats = 0 } 153 sd_child_payload(hbpath, job[SD_J_WS], payload, job[SD_J_LIMIT], beats, beat_epoch) 154 sys_exit(73) // unreachable safety net 155 return 73 156 } 157 // PARENT: record + an initial supervisor-side beat so the job is ALIVE at once. 158 job[SD_J_PID] = pid 159 job[SD_J_STATE] = SD_RUNNING 160 hb_beat_at(hbpath, job[SD_J_WS], beat_epoch, 0, sd_getpid()) 161 return pid 162} 163 164// NON-BLOCKING liveness poll of one job via wait4(pid, &st, WNOHANG). 165// returns 0 = still running (not reaped) 166// 1 = exited OK (code 0) -> job.state = EXITED_OK 167// 2 = crashed (nonzero exit OR killed by signal) -> job.state = CRASHED 168// Decodes status with the nx_container idiom (lo = status & 0xFF; exited iff lo==0). 169func sd_poll(job: *i64) -> i64 { 170 let stp: *i64 = sys_mmap(16) as *i64 171 stp[0] = 0 172 let r: i64 = sys_wait4(job[SD_J_PID], stp, WNOHANG) 173 if r == 0 { return 0 } // WNOHANG: child not yet reapable = running 174 if r < 0 { return 0 } // no such child to reap right now 175 let status: i64 = stp[0] 176 let lo: i64 = status & 0xFF 177 if lo == 0 { 178 let code: i64 = (status >> 8) & 0xFF 179 job[SD_J_EXIT] = code 180 if code == 0 { job[SD_J_STATE] = SD_EXITED_OK; return 1 } 181 job[SD_J_STATE] = SD_CRASHED 182 return 2 183 } 184 // killed by a signal (e.g. our SIGKILL) -> crashed 185 job[SD_J_EXIT] = 0 - (lo & 0x7F) 186 job[SD_J_STATE] = SD_CRASHED 187 return 2 188} 189 190// THE M5 BIND: dispatch a READY rung (M4) -> spawn it as a supervised job (M5). Maps the 191// leased business-key id to a numeric ws lane via id_to_ws (the heartbeat channel is keyed 192// by integer ws-id; the gate supplies the mapping for its hermetic fixture). Returns the 193// child pid (>0) on success, or -1 if nothing was dispatchable. 194func sd_dispatch_and_supervise(cx: *i64, dir: *u8, hbpath: *u8, job: *i64, 195 ws_id: i64, payload: i64, beat_epoch: i64, 196 id_out: *u8, fd_out: *i64) -> i64 { 197 let fd: i64 = dl_dispatch(cx, dir, id_out, fd_out) 198 if fd < 0 { return 0 - 1 } // nothing ready / all leased 199 job[SD_J_WS] = ws_id 200 let pid: i64 = sd_spawn_supervised(hbpath, job, payload, beat_epoch) 201 return pid 202} 203 204// THE RESTART DECISION (inlined supervisor policy + M1 evidence). Reads hbm_scan over the 205// single ws-id at the injected `now`/`thresh`: 206// - STALLED (missing heartbeat past threshold) AND restarts < budget -> RESPAWN (a fresh 207// supervised run with a fresh beat), restarts++, return ACTION_RESTART. 208// - STALLED AND budget exhausted -> mark ESCALATED, return ACTION_ESCALATE (no respawn). 209// - NOT stalled (alive, or it exited cleanly = EXITED_OK so it stopped beating on PURPOSE) 210// -> return ACTION_NONE. This is the no-thrash neg-control: a normally-completed job is 211// NEVER restarted. A naive M5 that respawns on "pid not alive" would WRONGLY restart the 212// clean job here (and FAIL the gate's T6). 213// `respawn_payload` selects the payload for the restarted run (the gate restarts with a 214// clean payload to prove the SECOND supervised run comes up ALIVE). 215func sd_supervise_tick(hbpath: *u8, now: i64, thresh: i64, job: *i64, 216 respawn_payload: i64, respawn_epoch: i64) -> i64 { 217 // If the job exited cleanly on its own, it is DONE -- never restart (the no-thrash law). 218 if job[SD_J_STATE] == SD_EXITED_OK { return SD_ACTION_NONE } 219 220 let ids: *i64 = sys_mmap(8) as *i64 221 ids[0] = job[SD_J_WS] 222 let vd: *i64 = sys_mmap(8) as *i64 223 let outs: *i64 = sys_mmap(32) as *i64 224 hbm_scan(hbpath, now, thresh, ids, 1, vd, outs) 225 let verdict: i64 = vd[0] 226 227 if verdict == HB_STALLED { 228 if job[SD_J_RESTARTS] < SD_MAXRESTART { 229 // RESTART / reassign: a fresh supervised run + a fresh beat clears the stall. 230 job[SD_J_RESTARTS] = job[SD_J_RESTARTS] + 1 231 sd_spawn_supervised(hbpath, job, respawn_payload, respawn_epoch) 232 return SD_ACTION_RESTART 233 } 234 // budget exhausted -> ESCALATE (bounded: no infinite restart loop / car-alarm storm) 235 job[SD_J_STATE] = SD_ESCALATED 236 return SD_ACTION_ESCALATE 237 } 238 // ALIVE (or unknown) -> nothing to do. 239 return SD_ACTION_NONE 240} 241 242// status emit: ONE buffer -> ONE locked fa_appendz (the cn_emit_* torn-line discipline). 243// "SUPDISP verb=<v> ws=<n> pid=<n> restarts=<n> action=<n> verdict=GREEN|RED" 244func sd_emit(path: *u8, verb: *u8, ws: i64, pid: i64, restarts: i64, action: i64, 245 ok: i64) -> i64 { 246 let buf: *u8 = sys_mmap(SD_REC_CAP + 16) 247 var o: i64 = 0 248 o = fa_cat(buf, o, "SUPDISP verb=\x00" as *u8) 249 o = fa_cat(buf, o, verb) 250 o = fa_cat(buf, o, " ws=\x00" as *u8) 251 o = fa_catn(buf, o, ws) 252 o = fa_cat(buf, o, " pid=\x00" as *u8) 253 o = fa_catn(buf, o, pid) 254 o = fa_cat(buf, o, " restarts=\x00" as *u8) 255 o = fa_catn(buf, o, restarts) 256 o = fa_cat(buf, o, " action=\x00" as *u8) 257 o = fa_catn(buf, o, action) 258 o = fa_cat(buf, o, " verdict=\x00" as *u8) 259 if ok == 1 { o = fa_cat(buf, o, "GREEN\x00" as *u8) } else { o = fa_cat(buf, o, "RED\x00" as *u8) } 260 buf[o] = 0 as u8 261 return fa_appendz(path, buf, SD_REC_CAP) 262} 263 264func sd_p(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 265func sd_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 } 266 267// CLI probe (NOT a daemon): self-author one supervised CLEAN job over a hermetic scratch 268// queue+heartbeat channel, poll it to completion, emit a status row. Proves the M5 organ 269// is live end-to-end on its own (the gate is the real referee). Time is real wall-clock 270// here (a smoke), but the GATE injects time for determinism. 271func main() -> i64 { 272 sd_p("=== nx_supervised_dispatch (M5) self-probe: spawn a supervised CLEAN job, poll, emit ===\n\x00" as *u8) 273 let ms: i64 = sys_now_realtime_ms() 274 let hb: *u8 = sys_mmap(96) 275 var o: i64 = 0 276 o = fa_cat(hb, o, "/tmp/sd_selfprobe_\x00" as *u8) 277 o = fa_catn(hb, o, ms) 278 o = fa_cat(hb, o, ".log\x00" as *u8) 279 hb[o] = 0 as u8 280 281 let job: *i64 = sd_job_new() 282 job[SD_J_WS] = 1 283 let epoch: i64 = sys_now_realtime_sec() 284 let pid: i64 = sd_spawn_supervised(hb, job, SD_PAY_CLEAN, epoch) 285 sd_p(" spawned supervised pid=\x00" as *u8); sd_n(pid) 286 sd_p(" limit_as=\x00" as *u8); sd_n(job[SD_J_LIMIT]); sd_p("\n\x00" as *u8) 287 288 // poll to completion (bounded loop; the clean payload exits promptly) 289 var guard: i64 = 0 290 var done: i64 = 0 291 while done == 0 { 292 let r: i64 = sd_poll(job) 293 if r != 0 { done = 1 } 294 guard = guard + 1 295 if guard > SD_MAGIC_100000 { done = 1 } 296 if done == 0 { sys_sleep_ms(2) } 297 } 298 sd_p(" job state=\x00" as *u8); sd_n(job[SD_J_STATE]) 299 sd_p(" exit=\x00" as *u8); sd_n(job[SD_J_EXIT]); sd_p("\n\x00" as *u8) 300 sd_emit("knowledge/status/supervised_dispatch.log\x00" as *u8, "selfprobe\x00" as *u8, 301 job[SD_J_WS], pid, job[SD_J_RESTARTS], SD_ACTION_NONE, 1) 302 sd_p(" SUPERVISED-DISPATCH self-probe: filed (knowledge/status/supervised_dispatch.log)\n\x00" as *u8) 303 sys_exit(0) 304 return 0 305}