nx_mesh_hostagent.nx source
↩ module page · 198 lines · 8859 B
1// nx_mesh_hostagent.nx -- WORKER MESH: the AUTONOMOUS host-agent that makes autoscale HANDS-OFF. It runs on the
2// worker host, and each tick it: reads DEMAND, PROBES the worker for warmth (bounded, ~1s), DECIDES via the autoscale
3// policy (SPAWN/SERVE/HOLD/REAP), and ACTUATES -- fork+execve the worker on SPAWN, nx_kill+wait4 on REAP. No human in
4// the loop: the organ physically starts and stops the worker process as demand comes and goes. On NishiOS this forks
5// the container natively; the exact same fork/kill lifecycle is exercised here against a real child process.
6//
7// CLI: (no args) -> self-test GATE (the decision policy)
8// demo <workerpath> <port> -> run the autonomous cold->SPAWN->SERVE->(idle)->REAP->cold cycle, forking and
9// killing a REAL worker child on a built-in demand schedule (proves the loop)
10// Actions: 0 HOLD 1 SPAWN 2 SERVE 3 REAP. license_tier: ORIGINAL
11import "nx_syscalls.nx"
12import "nx_connect.nx" // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
13import "nx_runtime.nx"
14import "nx_http_client.nx" // nx_http_client_sockaddr_ipv4 + nx_http_client_build_request (probe)
15const HA_MAGIC_2048: i64 = 2048
16const HA_MAGIC_2000: i64 = 2000
17const HA_MAGIC_4096: i64 = 4096
18const HA_MAGIC_4095: i64 = 4095
19
20const HA_HOLD: i64 = 0
21const HA_SPAWN: i64 = 1
22const HA_SERVE: i64 = 2
23const HA_REAP: i64 = 3
24
25func ha_p(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
26func ha_pn(v: i64) -> i64 {
27 let bb: *u8 = sys_mmap(28)
28 var m: i64 = v
29 if m < 0 { sys_write(1, "-" as *u8, 1); m = 0 - m }
30 let tt: *u8 = sys_mmap(28)
31 var k: i64 = 0
32 if m == 0 { tt[0] = 48 as u8; k = 1 }
33 while m > 0 { tt[k] = (48 + (m - (m/10)*10)) as u8; m = m / 10; k = k + 1 }
34 var i: i64 = 0
35 while i < k { bb[i] = tt[k-1-i]; i = i + 1 }
36 sys_write(1, bb, k)
37 return 0
38}
39func ha_atoi(s: *u8) -> i64 { var v: i64 = 0; var i: i64 = 0; while s[i] != (0 as u8) { let c: i64 = s[i] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 } return v }
40func ha_streq(a: *u8, b: *u8) -> i64 { var i: i64 = 0; while b[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 } if a[i] != (0 as u8) { return 0 } return 1 }
41func ha_action_name(a: i64) -> *u8 {
42 if a == HA_HOLD { return "HOLD" as *u8 }
43 if a == HA_SPAWN { return "SPAWN" as *u8 }
44 if a == HA_SERVE { return "SERVE" as *u8 }
45 return "REAP" as *u8
46}
47
48// ---- the policy (identical to nx_mesh_autoscale am_decide) ----
49func ha_decide(warm: i64, demand: i64, idle: i64, ttl: i64) -> i64 {
50 if demand == 1 { if warm == 0 { return HA_SPAWN } return HA_SERVE }
51 if warm == 1 { if idle >= ttl { return HA_REAP } return HA_HOLD }
52 return HA_HOLD
53}
54
55// ---- bounded warmth probe (non-blocking connect + poll) on 127.0.0.1:port ----
56func ha_pollfd(pfd: *u8, fd: i64, events: i64) -> i64 {
57 pfd[0]=(fd&0xff) as u8; pfd[1]=((fd>>8)&0xff) as u8; pfd[2]=((fd>>16)&0xff) as u8; pfd[3]=((fd>>24)&0xff) as u8
58 pfd[4]=(events&0xff) as u8; pfd[5]=((events>>8)&0xff) as u8; pfd[6]=0 as u8; pfd[7]=0 as u8
59 return 0
60}
61func ha_resp_2xx(buf: *u8, n: i64) -> i64 {
62 var i: i64 = 0
63 while i + 3 < n {
64 if buf[i]==(50 as u8) { if buf[i+1]==(48 as u8) { if buf[i+2]==(48 as u8) { return 1 } } }
65 if buf[i]==(10 as u8) { i = n }
66 i = i + 1
67 }
68 return 0
69}
70func ha_probe(port: i64) -> i64 {
71 let sa: *u8 = sys_mmap(16)
72 nx_http_client_sockaddr_ipv4(sa, 127, 0, 0, 1, port)
73 let fd: i64 = sys_socket(AF_INET, SOCK_STREAM | HA_MAGIC_2048, 0)
74 if fd < 0 { return 0 }
75 nx_connect_bounded(fd, sa, 16, NX_CONN_DEFAULT_MS)
76 let pfd: *u8 = sys_mmap(8)
77 ha_pollfd(pfd, fd, 4)
78 let pn: i64 = sys_poll(pfd, 1, 1000)
79 if pn <= 0 { sys_close(fd); return 0 }
80 let rev: i64 = (pfd[6] as i64) | ((pfd[7] as i64) << 8)
81 if (rev & 8) != 0 { sys_close(fd); return 0 }
82 if (rev & 16) != 0 { sys_close(fd); return 0 }
83 let req: *u8 = sys_mmap(256)
84 let rl: i64 = nx_http_client_build_request("/v1/models" as *u8, 10, "worker" as *u8, 6, req)
85 sys_write(fd, req, rl)
86 ha_pollfd(pfd, fd, 1)
87 let pn2: i64 = sys_poll(pfd, 1, HA_MAGIC_2000)
88 if pn2 <= 0 { sys_close(fd); return 0 }
89 let out: *u8 = sys_mmap(HA_MAGIC_4096)
90 let got: i64 = sys_read(fd, out, HA_MAGIC_4095)
91 sys_close(fd)
92 if got <= 0 { return 0 }
93 return ha_resp_2xx(out, got)
94}
95
96func ha_sleep_ms(ms: i64) -> i64 { sys_poll(0 as *u8, 0, ms); return 0 }
97
98// ---- ACTUATION: fork+execve the worker; nx_kill+wait4 to reap ----
99func ha_spawn(path: *u8, portstr: *u8) -> i64 {
100 let pid: i64 = sys_fork()
101 if pid == 0 {
102 let av: *i64 = sys_mmap(64) as *i64
103 av[0] = path as i64; av[1] = portstr as i64; av[2] = 0
104 sys_execve(path, av, 0 as *i64)
105 sys_exit(127) // only if execve failed
106 }
107 return pid
108}
109func ha_reap(pid: i64) -> i64 {
110 nx_kill(pid, 9) // SIGKILL -- guaranteed release of VRAM/port
111 let st: *i64 = sys_mmap(8) as *i64
112 sys_wait4(pid, st, 0) // reap the zombie
113 return 0
114}
115
116func ha_gate() -> i64 {
117 ha_p("=== nx_mesh_hostagent: autonomous autoscale host-agent (fork worker on demand, reap on idle) ===\n" as *u8)
118 var t1: i64 = 0
119 if ha_decide(0,1,0,2) == HA_SPAWN { t1 = 1 }
120 var t2: i64 = 0
121 if ha_decide(1,1,0,2) == HA_SERVE { t2 = 1 }
122 var t3: i64 = 0
123 if ha_decide(1,0,3,2) == HA_REAP { t3 = 1 }
124 var t4: i64 = 0
125 if ha_decide(1,0,1,2) == HA_HOLD { t4 = 1 }
126 var t5: i64 = 0
127 if ha_decide(0,0,9,2) == HA_HOLD { t5 = 1 }
128 ha_p(" T1 cold+demand->SPAWN: " as *u8); ha_pn(t1)
129 ha_p(" | T2 warm+demand->SERVE: " as *u8); ha_pn(t2)
130 ha_p(" | T3 warm+idle->REAP: " as *u8); ha_pn(t3)
131 ha_p(" | T4 warm+fresh->HOLD: " as *u8); ha_pn(t4)
132 ha_p(" | T5 cold+nodemand->HOLD: " as *u8); ha_pn(t5); ha_p("\n" as *u8)
133 var pass: i64 = 0
134 if t1==1 { if t2==1 { if t3==1 { if t4==1 { if t5==1 { pass = 1 } } } } }
135 if pass == 1 { ha_p("MESHHOSTAGENTGATE verdict=GREEN (policy correct)\n" as *u8); return 0 }
136 ha_p("MESHHOSTAGENTGATE verdict=RED\n" as *u8)
137 return 1
138}
139
140// autonomous demo: demand high for the first 4 ticks then gone; the agent forks + reaps a REAL worker child.
141func ha_demo(path: *u8, portstr: *u8) -> i64 {
142 let port: i64 = ha_atoi(portstr)
143 ha_p("=== autonomous host-agent loop (worker=" as *u8); ha_p(path); ha_p(" :" as *u8); ha_pn(port); ha_p(") ===\n" as *u8)
144 var worker_pid: i64 = 0
145 var idle: i64 = 0
146 let ttl: i64 = 2
147 var tick: i64 = 0
148 var spawned_pid: i64 = 0
149 var killed_pid: i64 = 0
150 while tick < 10 {
151 var demand: i64 = 0
152 if tick < 4 { demand = 1 }
153 var warm: i64 = 0
154 if worker_pid != 0 { warm = ha_probe(port) }
155 let act: i64 = ha_decide(warm, demand, idle, ttl)
156 ha_p(" tick " as *u8); ha_pn(tick); ha_p(": demand=" as *u8); ha_pn(demand); ha_p(" warm=" as *u8); ha_pn(warm); ha_p(" pid=" as *u8); ha_pn(worker_pid); ha_p(" idle=" as *u8); ha_pn(idle); ha_p(" -> " as *u8); ha_p(ha_action_name(act))
157 if act == HA_SPAWN {
158 worker_pid = ha_spawn(path, portstr)
159 spawned_pid = worker_pid
160 idle = 0
161 ha_p(" (forked worker pid=" as *u8); ha_pn(worker_pid); ha_p(")" as *u8)
162 ha_sleep_ms(600) // let it bind before the next probe
163 }
164 if act == HA_SERVE { idle = 0 }
165 if act == HA_HOLD { if demand == 0 { if warm == 1 { idle = idle + 1 } } }
166 if act == HA_REAP {
167 killed_pid = worker_pid
168 ha_reap(worker_pid)
169 ha_p(" (killed worker pid=" as *u8); ha_pn(worker_pid); ha_p(")" as *u8)
170 worker_pid = 0
171 idle = 0
172 }
173 ha_p("\n" as *u8)
174 ha_sleep_ms(500)
175 tick = tick + 1
176 }
177 // safety: reap if still running
178 if worker_pid != 0 { ha_reap(worker_pid); killed_pid = worker_pid }
179 ha_p("SUMMARY: autonomously SPAWNED pid=" as *u8); ha_pn(spawned_pid); ha_p(" then REAPED pid=" as *u8); ha_pn(killed_pid)
180 var ok: i64 = 0
181 if spawned_pid > 0 { if killed_pid == spawned_pid { ok = 1 } }
182 if ok == 1 { ha_p(" -- HANDS-OFF CYCLE PROVEN (probe went 0->1->0 as the agent started+stopped the worker)\n" as *u8); return 0 }
183 ha_p(" -- INCOMPLETE\n" as *u8)
184 return 1
185}
186
187func main(argc: i64, argv: *i64) -> i64 {
188 if argc >= 2 {
189 let cmd: *u8 = argv[1] as *u8
190 if ha_streq(cmd, "demo" as *u8) == 1 {
191 if argc < 4 { ha_p("usage: nx_mesh_hostagent demo <workerpath> <port>\n" as *u8); return 2 }
192 return ha_demo(argv[2] as *u8, argv[3] as *u8)
193 }
194 ha_p("usage: nx_mesh_hostagent demo <workerpath> <port> (no args = gate)\n" as *u8)
195 return 2
196 }
197 return ha_gate()
198}