nx_worker_dispatch.nx source
↩ module page · 262 lines · 13755 B
1// nx_worker_dispatch.nx -- WORKER MESH keystone: capability-scoped job dispatch across a sovereign GPU/compute mesh.
2//
3// The operator architecture: the NAS runs the orchestration container; it dispatches jobs BY API to compute WORKERS
4// (this laptop's RTX 5080 for image gen, the west server's dual 3090s for video, the cloud for other work). Each job
5// runs under a SCOPED CAPABILITY -- MCP-client style, it may ONLY touch the workers + storage it was granted, "trusted
6// to do work" enforced deny-by-default. Results are pushed to consumable storage (nishifamily.com/gallery).
7//
8// This organ is the DISPATCH BRAIN + the capability enforcement. It:
9// * ROUTES a job by type -> worker class (image->gpu-image / video->gpu-video / llm|embed|other->cloud)
10// * ENFORCES the capability: a job is AUTHORIZED only if its caps grant BOTH the routed worker AND the storage
11// target (missing either -> DENY). This is the "real MCP-type capability to access trusted resources".
12// * RESOLVES the worker endpoint (ip:port:path) + the storage target (gallery ingest).
13// * ACTUATES: for an authorized job it can CALL the live worker over sovereign TCP (nx_http_client) to prove the
14// authorized call reaches real compute -- the first leg of NAS -> worker -> gallery.
15//
16// CLI: (no args) -> self-test GATE (routing + capability enforcement + neg-controls)
17// probe <worker-class> -> live: GET the worker's /v1/models, report reachability
18// dispatch <type> <storage> <caps> -> authorize (deny-by-default) + probe the routed worker + report
19// Worker classes: gpu-image gpu-video cloud. caps = comma list e.g. "gpu-image,gallery".
20// NO fake greens: the gate mechanically proves the capability check DISCRIMINATES (grant->allow, missing->deny).
21// license_tier: ORIGINAL
22import "nx_syscalls.nx"
23import "nx_runtime.nx"
24import "nx_mesh_probe.nx" // mp_probe -- the shared BOUNDED probe (never hangs on a down worker)
25import "nx_swarm_endpoint_lib.nx" // role->addr SSOT (seq1485): the ADDRESS is data, not a literal
26const K_MAGIC_7861: i64 = 7861
27const K_MAGIC_5080: i64 = 5080
28
29func wd_w(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 }
30func wd_wn(fd: i64, v: i64) -> i64 {
31 let bb: *u8 = sys_mmap(28)
32 var m: i64 = v
33 if m < 0 { sys_write(fd, "-" as *u8, 1); m = 0 - m }
34 let tt: *u8 = sys_mmap(28)
35 var k: i64 = 0
36 if m == 0 { tt[0] = 48 as u8; k = 1 }
37 while m > 0 { tt[k] = (48 + (m - (m/10)*10)) as u8; m = m / 10; k = k + 1 }
38 var i: i64 = 0
39 while i < k { bb[i] = tt[k-1-i]; i = i + 1 }
40 sys_write(fd, bb, k)
41 return 0
42}
43func wd_p(s: *u8) -> i64 { return wd_w(1, s) }
44func wd_pn(v: i64) -> i64 { return wd_wn(1, v) }
45func wd_streq(a: *u8, b: *u8) -> i64 {
46 var i: i64 = 0
47 while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 }
48 if b[i] != (0 as u8) { return 0 }
49 return 1
50}
51
52// ---- routing: job type -> worker id (0 gpu-image, 1 gpu-video, 2 cloud, -1 unknown) ----
53func wd_route(jtype: *u8) -> i64 {
54 if wd_streq(jtype, "image" as *u8) == 1 { return 0 }
55 if wd_streq(jtype, "img" as *u8) == 1 { return 0 }
56 if wd_streq(jtype, "video" as *u8) == 1 { return 1 }
57 if wd_streq(jtype, "vid" as *u8) == 1 { return 1 }
58 if wd_streq(jtype, "llm" as *u8) == 1 { return 2 }
59 if wd_streq(jtype, "embed" as *u8) == 1 { return 2 }
60 if wd_streq(jtype, "text" as *u8) == 1 { return 2 }
61 return 0 - 1
62}
63func wd_worker_name(id: i64) -> *u8 {
64 if id == 0 { return "gpu-image" as *u8 }
65 if id == 1 { return "gpu-video" as *u8 }
66 if id == 2 { return "cloud" as *u8 }
67 return "unknown" as *u8
68}
69// "<label><addr>" with the ADDRESS resolved from the endpoint table (seq1485). The label is a human
70// string that does not drift; the address is the part DHCP moves, so only the address is data.
71// Fail-closed: an unresolvable role renders NO address rather than a stale one.
72func wd_where_role(role: *u8, label: *u8) -> *u8 {
73 let a: *u8 = se_addr(role)
74 if (a as i64) == 0 { return "unknown (role not in endpoint table)" as *u8 }
75 let o: *u8 = sys_mmap(256)
76 var k: i64 = se_put(o, 0, label)
77 k = se_put(o, k, a)
78 o[k] = 0 as u8
79 return o
80}
81
82func wd_worker_where(id: i64) -> *u8 {
83 if id == 0 { return wd_where_role("gpu-image" as *u8, "laptop RTX 5080 @ " as *u8) }
84 if id == 1 { return wd_where_role("gpu-video" as *u8, "west server dual 3090 @ " as *u8) }
85 if id == 2 { return "cloud inference endpoint" as *u8 }
86 return "unknown" as *u8
87}
88// fill ip octets + port for a worker id (image/video are LAN; cloud is resolved elsewhere). returns 1 ok / 0 no-lan-endpoint
89func wd_worker_addr(id: i64, ip: *i64, portbox: *i64) -> i64 {
90 if id == 0 { ip[0]=192; ip[1]=168; ip[2]=8; ip[3]=193; portbox[0]=K_MAGIC_7861; return 1 } // laptop K_MAGIC_5080
91 if id == 1 { ip[0]=10; ip[1]=0; ip[2]=4; ip[3]=13; portbox[0]=K_MAGIC_7861; return 1 } // west 3090s
92 return 0 // cloud: not a fixed LAN ipv4 (resolved via the cloud connector)
93}
94func wd_worker_id_by_name(nm: *u8) -> i64 {
95 if wd_streq(nm, "gpu-image" as *u8) == 1 { return 0 }
96 if wd_streq(nm, "gpu-video" as *u8) == 1 { return 1 }
97 if wd_streq(nm, "cloud" as *u8) == 1 { return 2 }
98 return 0 - 1
99}
100
101// ---- capability: is `token` a comma-delimited field of `csv`? (walk fields explicitly, boundary = ',' or NUL) ----
102func wd_field_eq(csv: *u8, s: i64, e: i64, token: *u8, tl: i64) -> i64 {
103 if (e - s) != tl { return 0 }
104 var i: i64 = 0
105 while i < tl { if csv[s + i] != token[i] { return 0 } i = i + 1 }
106 return 1
107}
108func wd_caps_has(csv: *u8, token: *u8) -> i64 {
109 var tl: i64 = 0
110 while token[tl] != (0 as u8) { tl = tl + 1 }
111 if tl == 0 { return 0 }
112 var p: i64 = 0
113 var fs: i64 = 0
114 var done: i64 = 0
115 while done == 0 {
116 let ch: i64 = csv[p] as i64
117 if ch == 44 {
118 if wd_field_eq(csv, fs, p, token, tl) == 1 { return 1 }
119 fs = p + 1
120 }
121 if ch == 0 {
122 if wd_field_eq(csv, fs, p, token, tl) == 1 { return 1 }
123 done = 1
124 }
125 p = p + 1
126 }
127 return 0
128}
129
130// ---- authorize a job: returns worker id if AUTHORIZED, else negative code.
131// -1 unknown type; -2 caps missing the routed worker grant; -3 caps missing the storage grant. ----
132func wd_authorize(jtype: *u8, storage: *u8, caps: *u8) -> i64 {
133 let id: i64 = wd_route(jtype)
134 if id < 0 { return 0 - 1 }
135 let wn: *u8 = wd_worker_name(id)
136 if wd_caps_has(caps, wn) == 0 { return 0 - 2 }
137 if wd_caps_has(caps, storage) == 0 { return 0 - 3 }
138 return id
139}
140
141// storage target -> the consumable location (data-driven; gallery is the default push surface)
142func wd_storage_where(storage: *u8) -> *u8 {
143 if wd_streq(storage, "gallery" as *u8) == 1 { return "nishifamily.com/gallery (nx_asset_ingest -> gallery catalog)" as *u8 }
144 if wd_streq(storage, "cas" as *u8) == 1 { return "content-addressed registry (nx_container_registry)" as *u8 }
145 return "unregistered storage target" as *u8
146}
147
148// ---- live actuation: GET the worker's /v1/models, report 2xx reachability ----
149func wd_resp_2xx(buf: *u8, n: i64) -> i64 {
150 // look for " 200" or "HTTP/1.1 2" in the first line
151 var i: i64 = 0
152 while i + 3 < n {
153 if buf[i] == (50 as u8) { if buf[i+1] == (48 as u8) { if buf[i+2] == (48 as u8) { return 1 } } } // "200"
154 if buf[i] == (10 as u8) { i = n } // stop at first newline (status line only)
155 i = i + 1
156 }
157 return 0
158}
159func wd_probe(id: i64) -> i64 {
160 let ip: *i64 = sys_mmap(32) as *i64
161 let portbox: *i64 = sys_mmap(8) as *i64
162 if wd_worker_addr(id, ip, portbox) == 0 { wd_p(" (cloud worker: no LAN probe; use the cloud connector)\n" as *u8); return 0 - 9 }
163 // BOUNDED probe (shared lib): a down worker is detected in ~1s instead of hanging -> the pipeline degrades gracefully.
164 if mp_probe(ip[0], ip[1], ip[2], ip[3], portbox[0]) == 1 { return 1 }
165 return 0 - 2
166}
167
168// ---- self-test GATE (no network) ----
169func wd_gate() -> i64 {
170 wd_p("=== nx_worker_dispatch: capability-scoped GPU/compute worker-mesh dispatch (NAS -> worker -> gallery) ===\n" as *u8)
171 // T1 routing
172 var t1: i64 = 0
173 if wd_route("image" as *u8) == 0 { if wd_route("video" as *u8) == 1 { if wd_route("llm" as *u8) == 2 { t1 = 1 } } }
174 // neg: unknown type -> -1
175 var neg1: i64 = 0
176 if wd_route("bogus-type-9Z" as *u8) == (0 - 1) { neg1 = 1 }
177 // T2 authorize OK: image job granted gpu-image + gallery -> worker id 0
178 var t2: i64 = 0
179 if wd_authorize("image" as *u8, "gallery" as *u8, "gpu-image,gallery" as *u8) == 0 { t2 = 1 }
180 // T3 DENY missing worker grant (caps has gallery but not gpu-image) -> -2
181 var t3: i64 = 0
182 if wd_authorize("image" as *u8, "gallery" as *u8, "gallery" as *u8) == (0 - 2) { t3 = 1 }
183 // T4 DENY missing storage grant (caps has gpu-image but not gallery) -> -3
184 var t4: i64 = 0
185 if wd_authorize("image" as *u8, "gallery" as *u8, "gpu-image" as *u8) == (0 - 3) { t4 = 1 }
186 // T5 video routes to gpu-video and needs the gpu-video grant
187 var t5: i64 = 0
188 if wd_authorize("video" as *u8, "gallery" as *u8, "gpu-video,gallery" as *u8) == 1 {
189 if wd_authorize("video" as *u8, "gallery" as *u8, "gpu-image,gallery" as *u8) == (0 - 2) { t5 = 1 }
190 }
191 // T6 worker addr resolves (laptop 5080)
192 var t6: i64 = 0
193 let ip: *i64 = sys_mmap(32) as *i64
194 let pb: *i64 = sys_mmap(8) as *i64
195 if wd_worker_addr(0, ip, pb) == 1 { if ip[0]==192 { if ip[3]==193 { if pb[0]==K_MAGIC_7861 { t6 = 1 } } } }
196 // neg2 deny-by-default: empty caps never authorizes
197 var neg2: i64 = 0
198 if wd_authorize("image" as *u8, "gallery" as *u8, "" as *u8) < 0 { neg2 = 1 }
199
200 wd_p(" T1 route image/video/llm->workers: " as *u8); wd_pn(t1)
201 wd_p(" | T2 authorize granted: " as *u8); wd_pn(t2)
202 wd_p(" | T3 deny missing-worker-cap: " as *u8); wd_pn(t3)
203 wd_p(" | T4 deny missing-storage-cap: " as *u8); wd_pn(t4)
204 wd_p(" | T5 video needs gpu-video grant: " as *u8); wd_pn(t5)
205 wd_p(" | T6 worker addr resolves: " as *u8); wd_pn(t6)
206 wd_p(" | neg1 unknown-type: " as *u8); wd_pn(neg1)
207 wd_p(" | neg2 deny-by-default: " as *u8); wd_pn(neg2); wd_p("\n" as *u8)
208
209 let sfd: i64 = sys_openat_wr("knowledge/status/worker_dispatch.tsv" as *u8, 0x1a4)
210 if sfd >= 0 {
211 wd_w(sfd, "# nx_worker_dispatch -- capability-scoped worker-mesh dispatch (image->5080 / video->3090 / other->cloud)\n" as *u8)
212 wd_w(sfd, "routing\t" as *u8); wd_wn(sfd, t1); wd_w(sfd, "\n" as *u8)
213 wd_w(sfd, "capability_enforced_deny_by_default\t" as *u8); wd_wn(sfd, t3); wd_w(sfd, "\n" as *u8)
214 sys_close(sfd)
215 }
216
217 var pass: i64 = 0
218 if t1==1 { if t2==1 { if t3==1 { if t4==1 { if t5==1 { if t6==1 { if neg1==1 { if neg2==1 { pass = 1 } } } } } } } }
219 if pass == 1 {
220 wd_p("WORKERDISPATCHGATE verdict=GREEN (routing correct; capability enforced deny-by-default; liar-killed)\n" as *u8)
221 return 0
222 }
223 wd_p("WORKERDISPATCHGATE verdict=RED (routing/capability/neg-control failed)\n" as *u8)
224 return 1
225}
226
227func main(argc: i64, argv: *i64) -> i64 {
228 if argc >= 2 {
229 let cmd: *u8 = argv[1] as *u8
230 if wd_streq(cmd, "probe" as *u8) == 1 {
231 if argc < 3 { wd_w(2, "usage: nx_worker_dispatch probe <worker-class>\n" as *u8); return 2 }
232 let id: i64 = wd_worker_id_by_name(argv[2] as *u8)
233 if id < 0 { wd_w(2, "unknown worker-class (gpu-image|gpu-video|cloud)\n" as *u8); return 3 }
234 wd_p("probe " as *u8); wd_p(wd_worker_name(id)); wd_p(" -> " as *u8); wd_p(wd_worker_where(id)); wd_p("\n" as *u8)
235 let r: i64 = wd_probe(id)
236 if r > 0 { wd_p(" WORKER REACHABLE (HTTP 200, " as *u8); wd_pn(r); wd_p(" bytes) -- live compute endpoint\n" as *u8); return 0 }
237 if r == (0 - 2) { wd_p(" UNREACHABLE (no connect/response) -- worker offline or not routed\n" as *u8); return 4 }
238 if r == (0 - 3) { wd_p(" reached but non-2xx\n" as *u8); return 5 }
239 return 6
240 }
241 if wd_streq(cmd, "dispatch" as *u8) == 1 {
242 if argc < 5 { wd_w(2, "usage: nx_worker_dispatch dispatch <type> <storage> <caps-csv>\n" as *u8); return 2 }
243 let jtype: *u8 = argv[2] as *u8
244 let storage: *u8 = argv[3] as *u8
245 let caps: *u8 = argv[4] as *u8
246 let id: i64 = wd_authorize(jtype, storage, caps)
247 wd_p("dispatch type=" as *u8); wd_p(jtype); wd_p(" storage=" as *u8); wd_p(storage); wd_p(" caps=" as *u8); wd_p(caps); wd_p("\n" as *u8)
248 if id == (0 - 1) { wd_p(" DENIED: unknown job type\n" as *u8); return 4 }
249 if id == (0 - 2) { wd_p(" DENIED: capability does not grant worker " as *u8); wd_p(wd_worker_name(wd_route(jtype))); wd_p(" (deny-by-default)\n" as *u8); return 5 }
250 if id == (0 - 3) { wd_p(" DENIED: capability does not grant storage " as *u8); wd_p(storage); wd_p(" (deny-by-default)\n" as *u8); return 6 }
251 wd_p(" AUTHORIZED -> worker " as *u8); wd_p(wd_worker_name(id)); wd_p(" (" as *u8); wd_p(wd_worker_where(id)); wd_p(")\n" as *u8)
252 wd_p(" push target: " as *u8); wd_p(wd_storage_where(storage)); wd_p("\n" as *u8)
253 let r: i64 = wd_probe(id)
254 if r > 0 { wd_p(" worker liveness: REACHABLE (HTTP 200) -- ready to run the job\n" as *u8); return 0 }
255 wd_p(" worker liveness: offline (start the worker, then re-dispatch)\n" as *u8)
256 return 0
257 }
258 wd_w(2, "usage: nx_worker_dispatch probe|dispatch (no args = self-test gate)\n" as *u8)
259 return 2
260 }
261 return wd_gate()
262}