`nx_wd_core.nx` source

↩ module page · 440 lines · 16519 B
1// nx_wd_core.nx -- DISPATCH-ACT CORE (mainless): the Warden's attempt loop
2// over the assignment queue (X-Q-003 rung 1; spec 2026-06-11-sclass-exceed-
3// backlog.md ADDENDUM Q3).  RACI: Warden verb = ATTEMPT/route.  This core
4// owns NO priority (PM's w column), NO sequencing policy (an_pick is the
5// Conductor's, imported), NO done-judging (nx_reconcile owns DONE flips on
6// MARK evidence -- this core RUNS reconcile and READS the queue back).
7//
8// THE BEAT (wd_beat), bounded to <=1 raise + <=1 attempt per call:
9//   pick = an_pick (highest-w runnable TODO)
10//   pick has ||EXEC=<organ> recipe  -> ATTEMPT: lane-build+run the organ
11//        (deadline-guarded, hang-proof), run nx_reconcile, re-read the queue;
12//        row DONE  -> ASSIGNRESULT verdict=GREEN
13//        row !DONE -> flip row RED (fail-loud, no livelock) + verdict=RED
14//   pick has no recipe -> RAISE: flip row TODO->NOVEL (tutor batch picks it
15//        up -- "a RAISED-HAND row instead of a silent stall") AND still
16//        attempt the best EXEC-covered runnable row so the beat does work.
17// Rung 2 (filed as its own queue row): covered-SHAPE routing via the warden
18// pattern classifier -> emitter lanes; this rung covers the EXEC-recipe class
19// (the X-EX-001 convention) only, and ASSIGNRESULT route= says which.
20//
21// Trust boundary (defensive at boundary law): EXEC basenames come from a
22// hand-curated TSV but are validated anyway -- [a-z0-9_] only, 1..63 chars,
23// so a recipe can never path-escape the lane's source resolution.
24// Queue rewrites change line length (TODO->NOVEL grows, TODO->RED shrinks);
25// sys_openat_wr has no O_TRUNC, so shrink pads trailing '\n' bytes -- an_load
26// and the line-walkers all skip empty lines (verified against their parsers).
27// license_tier: ORIGINAL
28
29import "nx_syscalls.nx"
30import "nx_itoa_lib.nx"   // shared MSB-first emitter (zero-alloc)
31import "nx_assign_core.nx"
32import "nx_guarded_run.nx"
33const WD_MAGIC_1048576: i64 = 1048576
34const WD_MAGIC_1048575: i64 = 1048575
35const WD_MAGIC_4096: i64 = 4096
36const WD_MAGIC_4095: i64 = 4095
37
38const WD_DEADLINE_DEFAULT: i64 = 600000   // ms; overridden by work_dispatcher.conf exec_deadline_ms=
39const WD_EXECCAP: i64 = 64
40
41// wcx block: wcx[0]=queue path  wcx[1]=results log path  wcx[2]=lane elf path
42// wcx[3]=reconcile elf path  wcx[4]=deadline ms  wcx[5]=open results fd
43// wcx[6]=auto-builder elf path (the SPEC route: X-Q-006 rung "rows that BUILD")
44func wd_newwcx(qp: *u8, respath: *u8, lane: *u8, rec: *u8, ab: *u8, dl: i64) -> *i64 {
  let w: *i64 = sys_mmap(64) as *i64
  w[0] = qp as i64
  w[1] = respath as i64
  w[2] = lane as i64
  w[3] = rec as i64
  w[4] = dl
  w[5] = 0 - 1
  w[6] = ab as i64
  return w
54}
55
56func 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 }
57// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
58// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
59// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
60// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
61func wd_wn(fd: i64, v: i64) -> i64 { nxi_fd(fd, v); return 0 }
62
63func wd_both(wcx: *i64, s: *u8) -> i64 {
  wd_w(1, s)
  let fd: i64 = wcx[5]
  if fd >= 0 { wd_w(fd, s) }
  return 0
68}
69
70func wd_bothn(wcx: *i64, v: i64) -> i64 {
  wd_wn(1, v)
  let fd: i64 = wcx[5]
  if fd >= 0 { wd_wn(fd, v) }
  return 0
75}
76
77// pattern-at: does buf[i..] start with pat (pl bytes)?  (rc_at shape)
78func wd_at(buf: *u8, n: i64, i: i64, pat: *u8, pl: i64) -> i64 {
  if i + pl > n { return 0 }
  var k: i64 = 0
  while k < pl { if buf[i + k] != pat[k] { return 0 } k = k + 1 }
  return 1
83}
84
85// extract ||EXEC=<basename> from a gate cell into out (cap WD_EXECCAP).
86// returns 1 valid recipe, 0 none-or-invalid.  Charset law: [a-z0-9_] only.
87func wd_exec_name(gate: *u8, out: *u8) -> i64 {
  var gl: i64 = 0
  while gate[gl] != (0 as u8) { gl = gl + 1 }
  var i: i64 = 0
  var at: i64 = 0 - 1
  while i < gl {
      if wd_at(gate, gl, i, "||EXEC=" as *u8, 7) == 1 { at = i + 7; i = gl } else { i = i + 1 }
  }
  if at < 0 { return 0 }
  var k: i64 = 0
  var go: i64 = 1
  while go == 1 {
      let ch: i64 = gate[at + k] as i64
      var stop: i64 = 0
      if ch == 0 { stop = 1 }
      if ch == 9 { stop = 1 }
      if ch == 124 { stop = 1 }
      if stop == 1 { go = 0 } else {
          var okc: i64 = 0
          if ch >= 97 { if ch <= 122 { okc = 1 } }
          if ch >= 48 { if ch <= 57 { okc = 1 } }
          if ch == 95 { okc = 1 }
          if okc == 0 { return 0 }
          if k >= WD_EXECCAP - 1 { return 0 }
          out[k] = ch as u8
          k = k + 1
      }
  }
  if k == 0 { return 0 }
  out[k] = 0 as u8
  return 1
118}
119
120const WD_SPECCAP: i64 = 192
121
122// extract ||SPEC=<specfile> from a gate cell into out (cap WD_SPECCAP).
123// returns 1 valid, 0 none-or-invalid.  Trust boundary: the path must live
124// under knowledge/specs/ (no traversal -- ".." rejected), charset
125// [a-z0-9_/.-] only, so a row can never route the builder outside the tree.
126func wd_spec_name(gate: *u8, out: *u8) -> i64 {
  var gl: i64 = 0
  while gate[gl] != (0 as u8) { gl = gl + 1 }
  var i: i64 = 0
  var at: i64 = 0 - 1
  while i < gl {
      if wd_at(gate, gl, i, "||SPEC=" as *u8, 7) == 1 { at = i + 7; i = gl } else { i = i + 1 }
  }
  if at < 0 { return 0 }
  var k: i64 = 0
  var go: i64 = 1
  while go == 1 {
      let ch: i64 = gate[at + k] as i64
      var stop: i64 = 0
      if ch == 0 { stop = 1 }
      if ch == 9 { stop = 1 }
      if ch == 124 { stop = 1 }
      if stop == 1 { go = 0 } else {
          var okc: i64 = 0
          if ch >= 97 { if ch <= 122 { okc = 1 } }
          if ch >= 48 { if ch <= 57 { okc = 1 } }
          if ch == 95 { okc = 1 }
          if ch == 47 { okc = 1 }
          if ch == 46 { okc = 1 }
          if ch == 45 { okc = 1 }
          if okc == 0 { return 0 }
          if k >= WD_SPECCAP - 1 { return 0 }
          out[k] = ch as u8
          k = k + 1
      }
  }
  if k == 0 { return 0 }
  out[k] = 0 as u8
  if wd_at(out, k, 0, "knowledge/specs/" as *u8, 16) != 1 { return 0 }
  var j: i64 = 0
  while j + 1 < k {
      if out[j] == (46 as u8) { if out[j + 1] == (46 as u8) { return 0 } }
      j = j + 1
  }
  return 1
166}
167
168// is row r covered (carries a valid EXEC recipe or SPEC build)?
169func wd_covered(cx: *i64, r: i64) -> i64 {
  let tmp: *u8 = sys_mmap(WD_SPECCAP)
  if wd_exec_name(an_gate_at(cx, r), tmp) == 1 { return 1 }
  if wd_spec_name(an_gate_at(cx, r), tmp) == 1 { return 1 }
  return 0
174}
175
176// rewrite the queue file flipping row <id>'s status field (field index 5) to
177// newst.  Whole-file read -> patch into a second buffer -> write back; if the
178// patched file is SHORTER, pad with '\n' (no O_TRUNC in the syscall wrapper;
179// empty lines are invisible to every queue parser).  Returns 1 flipped, 0 not.
180func wd_flip_status(qp: *u8, id: *u8, newst: *u8) -> i64 {
  let buf: *u8 = sys_mmap(WD_MAGIC_1048576)
  let fd: i64 = sys_openat_rd(qp)
  if fd < 0 { return 0 }
  var n: i64 = 0
  var r: i64 = sys_read(fd, buf, WD_MAGIC_1048575)
  while r > 0 { n = n + r; r = sys_read(fd, buf + n, WD_MAGIC_1048575 - n) }
  sys_close(fd)
  var idl: i64 = 0
  while id[idl] != (0 as u8) { idl = idl + 1 }
  var nsl: i64 = 0
  while newst[nsl] != (0 as u8) { nsl = nsl + 1 }
  let out: *u8 = sys_mmap(WD_MAGIC_1048576 + 64)
  var o: i64 = 0
  var hit: i64 = 0
  var ls: i64 = 0
  var i: i64 = 0
  while i <= n {
      var eol: i64 = 0
      if i == n { eol = 1 } else { if buf[i] == (10 as u8) { eol = 1 } }
      if eol == 1 {
          var mrow: i64 = 0
          if wd_at(buf, n, ls, id, idl) == 1 {
              if ls + idl < n { if buf[ls + idl] == (9 as u8) { mrow = 1 } }
          }
          if mrow == 1 {
              if hit == 0 {
                  hit = 1
                  // copy fields 0..4 verbatim (5 tabs), swap field 5, copy rest
                  var tabs: i64 = 0
                  var j: i64 = ls
                  while tabs < 5 {
                      if j >= i { tabs = 5 } else {
                          out[o] = buf[j]
                          if buf[j] == (9 as u8) { tabs = tabs + 1 }
                          o = o + 1
                          j = j + 1
                      }
                  }
                  var k: i64 = 0
                  while k < nsl { out[o] = newst[k]; o = o + 1; k = k + 1 }
                  // skip the old status field content
                  var sk: i64 = 1
                  while sk == 1 {
                      if j >= i { sk = 0 } else {
                          if buf[j] == (9 as u8) { sk = 0 } else { j = j + 1 }
                      }
                  }
                  while j < i { out[o] = buf[j]; o = o + 1; j = j + 1 }
              } else {
                  var j2: i64 = ls
                  while j2 < i { out[o] = buf[j2]; o = o + 1; j2 = j2 + 1 }
              }
          } else {
              var j3: i64 = ls
              while j3 < i { out[o] = buf[j3]; o = o + 1; j3 = j3 + 1 }
          }
          if i < n { out[o] = 10 as u8; o = o + 1 }
          ls = i + 1
      }
      i = i + 1
  }
  if hit == 0 { return 0 }
  while o < n { out[o] = 10 as u8; o = o + 1 }
  let wf: i64 = sys_openat_wr(qp, 0x1a4)
  if wf < 0 { return 0 }
  var off: i64 = 0
  while off < o { let w2: i64 = sys_write(wf, out + off, o - off); if w2 <= 0 { off = o } else { off = off + w2 } }
  sys_close(wf)
  return 1
250}
251
252// the best runnable TODO row that is COVERED (EXEC recipe or SPEC build):
253// an_pick variant -- highest w, deps DONE.  -1 if none.
254func wd_pick_covered(cx: *i64) -> i64 {
  let wv: *i64 = cx[3] as *i64
  let st: *i64 = cx[4] as *i64
  let sz: *i64 = cx[5] as *i64
  var best: i64 = 0 - 1
  var r: i64 = 0
  while r < cx[0] {
      if st[r] == 84 {
          if wd_covered(cx, r) == 1 {
              if an_deps_done(cx, r) == 1 {
                  var take: i64 = 0
                  if best < 0 { take = 1 } else {
                      if wv[r] > wv[best] { take = 1 }
                      if wv[r] == wv[best] { if sz[r] < sz[best] { take = 1 } }
                  }
                  if take == 1 { best = r }
              }
          }
      }
      r = r + 1
  }
  return best
276}
277
278// spawn elf with ONE argument under the wcx deadline (hang-proof; composes
279// nx_guarded_run -- SIGKILL reap, rc=124 timeout convention).
280func wd_spawn2(wcx: *i64, elf: *u8, a1: *u8) -> i64 {
  let argv: *i64 = sys_mmap(32) as *i64
  argv[0] = elf as i64
  argv[1] = a1 as i64
  argv[2] = 0
  let envp: *i64 = sys_mmap(32) as *i64
  envp[0] = "PATH=/usr/bin:/bin" as *u8 as i64
  envp[1] = 0
  let neg: i64 = 0 - 1
  return nx_guarded_run(elf, argv, envp, wcx[4], neg, neg)
290}
291
292// one ASSIGNRESULT row (stdout + results log): the attempt audit trail.
293func wd_result(wcx: *i64, id: *u8, route: *u8, verdict: *u8, rc: i64) -> i64 {
  wd_both(wcx, "ASSIGNRESULT id=" as *u8)
  wd_both(wcx, id)
  wd_both(wcx, " route=" as *u8)
  wd_both(wcx, route)
  wd_both(wcx, " verdict=" as *u8)
  wd_both(wcx, verdict)
  wd_both(wcx, " rc=" as *u8)
  wd_bothn(wcx, rc)
  wd_both(wcx, " epoch=" as *u8)
  wd_bothn(wcx, sys_now_realtime_sec())
  wd_both(wcx, "\n" as *u8)
  return 0
306}
307
308// ATTEMPT one covered row end to end: run <elf a1> (the route's act), then
309// the status-keeper (nx_reconcile), then re-read the queue.  Returns 0 GREEN,
310// 1 RED.  Judgment is EVIDENCE ONLY -- the row is GREEN iff reconcile flipped
311// it DONE on its own MARK evaluation.  An act that exits 0 but never turns
312// its marker green is a LIAR and goes RED.
313func wd_attempt(wcx: *i64, id: *u8, elf: *u8, a1: *u8, route: *u8) -> i64 {
  let qp: *u8 = wcx[0] as *u8
  let echk: i64 = sys_openat_rd(elf)
  if echk < 0 {
      wd_result(wcx, id, route, "RED-act-elf-absent" as *u8, 127)
      return 1
  }
  sys_close(echk)
  let rc_run: i64 = wd_spawn2(wcx, elf, a1)
  let chk: i64 = sys_openat_rd(wcx[3] as *u8)
  if chk < 0 {
      wd_result(wcx, id, route, "RED-reconcile-elf-absent" as *u8, rc_run)
      return 1
  }
  sys_close(chk)
  wd_spawn2(wcx, wcx[3] as *u8, qp)
  let cx2: *i64 = an_newcx()
  let rows: i64 = an_load(qp, cx2)
  if rows <= 0 {
      wd_result(wcx, id, route, "RED-queue-reload-failed" as *u8, rc_run)
      return 1
  }
  let r2: i64 = an_find(cx2, id)
  if r2 >= 0 {
      let st: *i64 = cx2[4] as *i64
      if st[r2] == 68 {
          wd_result(wcx, id, route, "GREEN" as *u8, rc_run)
          return 0
      }
  }
  wd_flip_status(qp, id, "RED" as *u8)
  wd_result(wcx, id, route, "RED" as *u8, rc_run)
  return 1
346}
347
348// route one covered row by its kind: EXEC -> the durable lane rebuilds+runs
349// the recipe organ; SPEC -> the durable auto-builder authors+gates the
350// module from the spec file (classifier-routed emitter lanes: rows that
351// BUILD).  Returns wd_attempt's verdict; 1 if the row turned out uncovered
352// (caller bug -- fail loud).
353func wd_route_attempt(wcx: *i64, cx: *i64, r: i64) -> i64 {
  let idbuf: *u8 = sys_mmap(64)
  let pid: *u8 = an_id_at(cx, r)
  var ci: i64 = 0
  while pid[ci] != (0 as u8) { idbuf[ci] = pid[ci]; ci = ci + 1 }
  idbuf[ci] = 0 as u8
  let arg: *u8 = sys_mmap(WD_SPECCAP)
  if wd_exec_name(an_gate_at(cx, r), arg) == 1 {
      return wd_attempt(wcx, idbuf, wcx[2] as *u8, arg, "EXEC" as *u8)
  }
  if wd_spec_name(an_gate_at(cx, r), arg) == 1 {
      return wd_attempt(wcx, idbuf, wcx[6] as *u8, arg, "SPEC" as *u8)
  }
  wd_result(wcx, idbuf, "NONE" as *u8, "RED-uncovered-routed" as *u8, 0)
  return 1
368}
369
370// THE BEAT.  Returns 0 healthy (work done, raised, or idle), 1 attempt RED,
371// 101 queue missing.  Bounded: <=1 raise + <=1 attempt per call.
372func wd_beat(wcx: *i64) -> i64 {
  let qp: *u8 = wcx[0] as *u8
  let cx: *i64 = an_newcx()
  let rows: i64 = an_load(qp, cx)
  if rows <= 0 {
      wd_w(1, "ASSIGNRESULT id=NONE route=NONE verdict=RED-queue-missing\n" as *u8)
      return 101
  }
  let lfd: i64 = sys_openat_append(wcx[1] as *u8, 0x1a4)
  wcx[5] = lfd
  var bad: i64 = 0
  let p: i64 = an_pick(cx)
  if p < 0 {
      wd_result(wcx, "NONE-RUNNABLE" as *u8, "NONE" as *u8, "IDLE" as *u8, 0)
  } else {
      if wd_covered(cx, p) == 1 {
          bad = wd_route_attempt(wcx, cx, p)
      } else {
          let idbuf: *u8 = sys_mmap(64)
          let pid: *u8 = an_id_at(cx, p)
          var ci: i64 = 0
          while pid[ci] != (0 as u8) { idbuf[ci] = pid[ci]; ci = ci + 1 }
          idbuf[ci] = 0 as u8
          wd_flip_status(qp, idbuf, "NOVEL" as *u8)
          wd_result(wcx, idbuf, "NOVEL" as *u8, "RAISED" as *u8, 0)
          let pe: i64 = wd_pick_covered(cx)
          if pe >= 0 { bad = wd_route_attempt(wcx, cx, pe) }
      }
  }
  if lfd >= 0 { sys_close(lfd) }
  return bad
403}
404
405// deadline from knowledge/registry/work_dispatcher.conf (exec_deadline_ms=N);
406// bootstrap default when the conf or key is absent (config-hierarchy law).
407func wd_conf_deadline(path: *u8) -> i64 {
  let buf: *u8 = sys_mmap(WD_MAGIC_4096)
  let fd: i64 = sys_openat_rd(path)
  if fd < 0 { return WD_DEADLINE_DEFAULT }
  var n: i64 = 0
  var r: i64 = sys_read(fd, buf, WD_MAGIC_4095)
  while r > 0 { n = n + r; r = sys_read(fd, buf + n, WD_MAGIC_4095 - n) }
  sys_close(fd)
  let key: *u8 = "exec_deadline_ms=" as *u8
  var i: i64 = 0
  var at: i64 = 0 - 1
  while i < n {
      if wd_at(buf, n, i, key, 17) == 1 { at = i + 17; i = n } else { i = i + 1 }
  }
  if at < 0 { return WD_DEADLINE_DEFAULT }
  var v: i64 = 0
  var any: i64 = 0
  var go: i64 = 1
  while go == 1 {
      if at >= n { go = 0 } else {
          let ch: i64 = buf[at] as i64
          if ch >= 48 {
              if ch <= 57 {
                  v = v * 10 + (ch - 48)
                  any = 1
                  at = at + 1
              } else { go = 0 }
          } else { go = 0 }
      }
  }
  if any == 0 { return WD_DEADLINE_DEFAULT }
  if v < 1000 { return WD_DEADLINE_DEFAULT }
  return v
440}
nx_wd_core.nx source

`nx_wd_core.nx` source
