`nx_mesh_run.nx` source

↩ module page · 166 lines · 9348 B
1// nx_mesh_run.nx -- WORKER MESH CONDUCTOR: runs ONE job through the whole authorized pipeline by composing the REAL
2// mesh organs (not a re-implementation -- it fork+captures the proven elfs, gating each step on the next):
3//   1. AUTHORIZE  -> nx_worker_dispatch (capability-scoped, deny-by-default): may this job touch gpu-image + gallery?
4//   2. ROUTE      -> nx_mesh_lb (health-aware): pick a HEALTHY replica of the worker class (skip the down ones).
5//   3. EXECUTE    -> nx_mesh_gen (sovereign POST->decode->gallery): generate + publish, zero shell.
6// Each milestone gates the next: a capability denial stops before routing; no healthy replica stops before executing.
7// This is the mesh as ONE usable system (the operator's "the NAS dispatches jobs" = a single command), composing the
8// individually-proven organs. Run from the nishihost cwd where the organ elfs live.
9//
10// CLI:  run <prompt> <outpath> [caps]   -> the full pipeline (caps default gpu-image,gallery; pass a lesser set to see
11//                                          the capability gate DENY and stop the pipeline)
12//       (no args)                       -> self-test GATE (the substring gate that drives step transitions)
13// license_tier: ORIGINAL
14import "nx_syscalls.nx"
15import "nx_runtime.nx"
16import "nx_tool_run.nx"   // tr_run_capture (fork + argv + stdout capture)
17const K_MAGIC_4194304: i64 = 4194304
18
19func mr_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
20func mr_puts(s: *u8) -> i64 { sys_write(1, s, mr_slen(s)); return 0 }
21func mr_contains(buf: *u8, n: i64, needle: *u8) -> i64 {
  let m: i64 = mr_slen(needle)
  if m == 0 { return 0 }
  var i: i64 = 0
  while i + m <= n {
      var k: i64 = 0
      while k < m { if buf[i+k] != needle[k] { k = m + 9 } else { k = k + 1 } }
      if k == m { return 1 }
      i = i + 1
  }
  return 0
32}
33func mr_run(path: *u8, av: *i64, out: *u8, cap: i64, olen: *i64) -> i64 {
  av[0] = path as i64
  return tr_run_capture(path, av, out, cap, olen)
36}
37
38// ---- SSE emit: the mesh-service response body. A daemon writes these frames to the HTTP response as the pipeline
39// runs; here they go to stdout so the streaming is provable + browser-EventSource-shaped. Quotes = byte 34. ----
40func mr_wb(b: i64) -> i64 { let x: *u8 = sys_mmap(1); x[0] = b as u8; sys_write(1, x, 1); return 0 }
41func mr_sse1(event: *u8, status: *u8) -> i64 {
  mr_puts("event: " as *u8); mr_puts(event); mr_wb(10)
  mr_puts("data: {" as *u8); mr_wb(34); mr_puts("status" as *u8); mr_wb(34); mr_wb(58); mr_wb(34); mr_puts(status); mr_wb(34); mr_wb(125); mr_wb(10); mr_wb(10)
  return 0
45}
46func mr_sse2(event: *u8, status: *u8, key: *u8, val: *u8) -> i64 {
  mr_puts("event: " as *u8); mr_puts(event); mr_wb(10)
  mr_puts("data: {" as *u8); mr_wb(34); mr_puts("status" as *u8); mr_wb(34); mr_wb(58); mr_wb(34); mr_puts(status); mr_wb(34)
  mr_wb(44); mr_wb(34); mr_puts(key); mr_wb(34); mr_wb(58); mr_wb(34); mr_puts(val); mr_wb(34)
  mr_wb(125); mr_wb(10); mr_wb(10)
  return 0
52}
53// run the pipeline emitting SSE per REAL milestone. returns 0 (result) / nonzero (denied/no-worker/fail).
54func mr_sse_pipeline(prompt: *u8, outpath: *u8, caps: *u8) -> i64 {
  let out: *u8 = sys_mmap(K_MAGIC_4194304)
  let olen: *i64 = sys_mmap(16) as *i64
  let av: *i64 = sys_mmap(256) as *i64
  mr_puts("retry: 5000\n\n" as *u8)
  mr_sse2("accepted" as *u8, "accepted" as *u8, "prompt" as *u8, prompt)
  // AUTHORIZE
  av[1] = "dispatch" as *u8 as i64; av[2] = "image" as *u8 as i64; av[3] = "gallery" as *u8 as i64; av[4] = caps as i64; av[5] = 0
  mr_run("./nx_worker_dispatch.elf" as *u8, av, out, K_MAGIC_4194304, olen)
  if mr_contains(out, olen[0], "AUTHORIZED" as *u8) == 0 {
      mr_sse2("denied" as *u8, "denied" as *u8, "reason" as *u8, "capability not granted" as *u8)
      return 5
  }
  mr_sse1("authorized" as *u8, "authorized" as *u8)
  // ROUTE
  av[1] = "pick" as *u8 as i64; av[2] = "gpu-image" as *u8 as i64; av[3] = 0
  mr_run("./nx_mesh_lb.elf" as *u8, av, out, K_MAGIC_4194304, olen)
  if mr_contains(out, olen[0], "ROUTE" as *u8) == 0 {
      mr_sse2("error" as *u8, "unavailable" as *u8, "reason" as *u8, "no healthy replica" as *u8)
      return 6
  }
  mr_sse2("routed" as *u8, "routed" as *u8, "worker" as *u8, "gpu-image" as *u8)
  // EXECUTE
  mr_sse1("generating" as *u8, "generating" as *u8)
  av[1] = "gen" as *u8 as i64; av[2] = prompt as i64; av[3] = outpath as i64; av[4] = 0
  mr_run("./nx_mesh_gen.elf" as *u8, av, out, K_MAGIC_4194304, olen)
  if mr_contains(out, olen[0], "WROTE" as *u8) == 0 {
      mr_sse2("error" as *u8, "failed" as *u8, "reason" as *u8, "generation failed" as *u8)
      return 7
  }
  mr_sse2("result" as *u8, "done" as *u8, "url" as *u8, outpath)
  return 0
86}
87
88func mr_gate() -> i64 {
  mr_puts("=== nx_mesh_run: mesh conductor (authorize -> route -> execute, composes the real organs) ===\n" as *u8)
  var t1: i64 = 0
  if mr_contains("xx AUTHORIZED yy" as *u8, 16, "AUTHORIZED" as *u8) == 1 { t1 = 1 }
  var t2: i64 = 0
  if mr_contains("DENIED capability" as *u8, 17, "AUTHORIZED" as *u8) == 0 { t2 = 1 }
  var t3: i64 = 0
  if mr_contains("  ROUTE -> [0] laptop" as *u8, 21, "ROUTE" as *u8) == 1 { t3 = 1 }
  mr_puts("  T1 detects AUTHORIZED: " as *u8); if t1==1 { mr_puts("1" as *u8) } else { mr_puts("0" as *u8) }
  mr_puts("  | T2 rejects DENIED: " as *u8); if t2==1 { mr_puts("1" as *u8) } else { mr_puts("0" as *u8) }
  mr_puts("  | T3 detects ROUTE: " as *u8); if t3==1 { mr_puts("1" as *u8) } else { mr_puts("0" as *u8) }
  mr_puts("\n" as *u8)
  var pass: i64 = 0
  if t1==1 { if t2==1 { if t3==1 { pass = 1 } } }
  if pass == 1 { mr_puts("MESHRUNGATE verdict=GREEN (step-gating substring logic correct)\n" as *u8); return 0 }
  mr_puts("MESHRUNGATE verdict=RED\n" as *u8)
  return 1
105}
106
107func main(argc: i64, argv: *i64) -> i64 {
  if argc >= 2 {
      let cmd: *u8 = argv[1] as *u8
      // "run" ?
      if cmd[0]==(114 as u8) { if cmd[1]==(117 as u8) { if cmd[2]==(110 as u8) { if cmd[3]==(0 as u8) {
          if argc < 4 { mr_puts("usage: nx_mesh_run run <prompt> <outpath> [caps]\n" as *u8); return 2 }
          let prompt: *u8 = argv[2] as *u8
          let outpath: *u8 = argv[3] as *u8
          var caps: *u8 = "gpu-image,gallery" as *u8
          if argc >= 5 { caps = argv[4] as *u8 }
          let out: *u8 = sys_mmap(K_MAGIC_4194304)
          let olen: *i64 = sys_mmap(16) as *i64
          let av: *i64 = sys_mmap(256) as *i64
120
          // ---- milestone 1: AUTHORIZE (capability-scoped, deny-by-default) ----
          mr_puts("[mesh-run] 1/3 AUTHORIZE (capability=" as *u8); mr_puts(caps); mr_puts(")\n" as *u8)
          av[1] = "dispatch" as *u8 as i64; av[2] = "image" as *u8 as i64; av[3] = "gallery" as *u8 as i64; av[4] = caps as i64; av[5] = 0
          mr_run("./nx_worker_dispatch.elf" as *u8, av, out, K_MAGIC_4194304, olen)
          if mr_contains(out, olen[0], "AUTHORIZED" as *u8) == 0 {
              mr_puts("  DENIED by the capability gate -- pipeline STOPS (job never reached a worker)\n" as *u8)
              sys_write(1, out, olen[0])
              return 5
          }
          mr_puts("  authorized.\n" as *u8)
131
          // ---- milestone 2: ROUTE (health-aware LB) ----
          mr_puts("[mesh-run] 2/3 ROUTE (health-aware multi-worker)\n" as *u8)
          av[1] = "pick" as *u8 as i64; av[2] = "gpu-image" as *u8 as i64; av[3] = 0
          mr_run("./nx_mesh_lb.elf" as *u8, av, out, K_MAGIC_4194304, olen)
          if mr_contains(out, olen[0], "ROUTE" as *u8) == 0 {
              mr_puts("  NO HEALTHY REPLICA -- pipeline STOPS (would 503)\n" as *u8)
              sys_write(1, out, olen[0])
              return 6
          }
          sys_write(1, out, olen[0])
142
          // ---- milestone 3: EXECUTE (sovereign gen -> gallery) ----
          mr_puts("[mesh-run] 3/3 EXECUTE (sovereign POST->decode->gallery)\n" as *u8)
          av[1] = "gen" as *u8 as i64; av[2] = prompt as i64; av[3] = outpath as i64; av[4] = 0
          mr_run("./nx_mesh_gen.elf" as *u8, av, out, K_MAGIC_4194304, olen)
          sys_write(1, out, olen[0])
          if mr_contains(out, olen[0], "WROTE" as *u8) == 0 {
              mr_puts("  generation FAILED (worker offline?) -- pipeline STOPS\n" as *u8)
              return 7
          }
          mr_puts("[mesh-run] PIPELINE COMPLETE: authorized -> routed -> generated -> published " as *u8); mr_puts(outpath); mr_puts("\n" as *u8)
          return 0
      } } } }
      // sse <prompt> <outpath> [caps] -- run the pipeline, emit a live SSE stream of the REAL milestones (service body)
      if cmd[0]==(115 as u8) { if cmd[1]==(115 as u8) { if cmd[2]==(101 as u8) { if cmd[3]==(0 as u8) {
          if argc < 4 { mr_puts("usage: nx_mesh_run sse <prompt> <outpath> [caps]\n" as *u8); return 2 }
          var caps2: *u8 = "gpu-image,gallery" as *u8
          if argc >= 5 { caps2 = argv[4] as *u8 }
          return mr_sse_pipeline(argv[2] as *u8, argv[3] as *u8, caps2)
      } } } }
      mr_puts("usage: nx_mesh_run run|sse <prompt> <outpath> [caps]  (no args = gate)\n" as *u8)
      return 2
  }
  return mr_gate()
166}
nx_mesh_run.nx source

`nx_mesh_run.nx` source
