code wiki / _hdl_build / nx_queue_load.nx
`nx_queue_load.nx` source

↩ module page · 267 lines · 11515 B
1// nx_queue_load.nx -- IM-Q1: the PROGRAM'S OWN COORDINATION DATA becomes
2// first-class records on the sovereign store (operator 2026-06-11: "own our
3// information management -- a tsv file is not nishi; bottom up hardware rung
4// and each rung nishi").  The seg-store substrate is already hardware-up
5// (rungs 1-9, beats FTS5); this rung puts the QUEUE on it:
6//   assignment_queue.tsv rows -> key "aq:"+<row-id>, value = canonical record
7//     {kind: queue-row, id, row}.  SAME key, NEW VERSION on every change --
8//     `nx_store_query hist aq:X-Q-003` is the row's full status time-travel,
9//     something the bare TSV can never answer.  Idempotent: unchanged row =
10//     identical canonical bytes = dup_instore, never re-added.
11//   sponsor_decisions.tsv / planning_prefs.tsv / callouts.tsv lines ->
12//     content-CID line records (sd:/pp:/co: prefixes, nx_log_load pattern) --
13//     the sponsor's decisions + the training corpus gain provenance lineage.
14// The TSV files stay as the WRITE surface this rung (API stability); IM-Q2
15// (filed) moves writers store-first and derives the TSV as a render.
16// argv[1]=store prefix override, argv[2]=queue path override (gates).
17// Evidence: QUEUE-STORE row -> stdout + knowledge/status/infomgmt_store.log.
18// license_tier: ORIGINAL
19
20import "nx_syscalls.nx"
21import "nx_canon_cid.nx"
22import "nx_seg_store.nx"
23// Field 0 of a queue row is a short id like "X-Q-003". This bounds the SCAN for
24// the first tab and is a SEMANTIC rule, not a buffer capacity -- it is the only
25// bound left in this organ. Every buffer below is sized from its own input
26// (canon_encode_size for records, the line length for a row, the file length for
27// the reused line buffer), so no capacity constant has to be kept in sync with
28// anything, and nothing can be dropped because a guessed ceiling was too low.
29const QL_MAX_ID_LEN: i64 = 40
30
31func ql_p(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
32func ql_fp(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 }
33func ql_fn(fd: i64, v: i64) -> i64 { let bb: *u8=sys_mmap(28); var m: i64=v; if m<0{m=0-m; sys_write(fd,"-" as *u8,1)}; let t: *u8=sys_mmap(28); var k: i64=0; if m==0{t[0]=48;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(fd,bb,k); return 0 }
34
35func ql_memeq(a: *u8, b: *u8, n: i64) -> i64 {
  var i: i64 = 0
  while i < n {
      if a[i] != b[i] { return 0 }
      i = i + 1
  }
  return 1
42}
43
44// st: st[0]=writer st[1]=wdocs st[2]=segs st[3]=added st[4]=dup st[5]=scanned
45// st[6]=abort-rc st[7]=open handle
46const QL_CHUNK: i64 = 64
47
48func ql_flush(prefix: *u8, st: *i64) -> i64 {
  if st[1] == 0 { return 0 }
  let rc: i64 = ss_commit(prefix, st[0] as *i64, sys_now_us())
  if rc != 0 { st[6] = rc; return rc }
  st[0] = ss_begin() as i64
  st[1] = 0
  st[2] = st[2] + 1
  return 0
56}
57
58// add record if the store's latest version differs (or key absent); 0/neg
59func ql_put(prefix: *u8, st: *i64, kbuf: *u8, enc: *u8, el: i64) -> i64 {
  let pp: *i64 = sys_mmap(16) as *i64
  let ll: *i64 = sys_mmap(16) as *i64
  var have: i64 = 0
  let g: i64 = ss_hget(st[7] as *i64, kbuf, pp, ll)
  if g == 1 { if ll[0] == el { if ql_memeq(pp[0] as *u8, enc, el) == 1 { have = 1 } } }
  if have == 1 { st[4] = st[4] + 1; return 0 }
  if ss_add(st[0] as *i64, 1, kbuf, enc, el) != 0 { st[6] = 0 - 98; return 0 - 3 }
  st[3] = st[3] + 1
  st[1] = st[1] + 1
  if st[1] >= QL_CHUNK { if ql_flush(prefix, st) != 0 { return 0 - 4 } }
  return 0
71}
72
73// one queue row line [i,lend) -> aq:<id> versioned record
74func ql_row(prefix: *u8, st: *i64, b: *u8, i: i64, lend: i64) -> i64 {
  let llen: i64 = lend - i
  if llen <= 0 { return 0 }
  if b[i] == (35 as u8) { return 0 }
  // There is deliberately no length ceiling here. The buffers below are sized
  // from llen, so a cap would have nothing to protect. The 3800-byte guard this
  // replaces returned SUCCESS and fired BEFORE st[5] incremented, so an
  // over-long queue row was absent from the numerator AND the denominator --
  // nothing in this organ's output could ever reveal that a row was lost.
  // id = field 0 (to first tab); no tab = not a row
  var idl: i64 = 0
  var go: i64 = 1
  while go == 1 {
      if i + idl >= lend { return 0 } else {
          if b[i + idl] == (9 as u8) { go = 0 } else {
              idl = idl + 1
              // Semantic rejection, and it ANNOUNCES: a rejected line must never
              // be indistinguishable from a line that was not there.
              if idl > QL_MAX_ID_LEN { st[8] = st[8] + 1; return 0 }
          }
      }
  }
  if idl == 0 { return 0 }
  st[5] = st[5] + 1
  let idb: *u8 = sys_mmap(idl + 1)
  var t: i64 = 0
  while t < idl { idb[t] = b[i + t]; t = t + 1 }
  idb[idl] = 0 as u8
  let lbuf: *u8 = sys_mmap(llen + 1)
  t = 0
  while t < llen { lbuf[t] = b[i + t]; t = t + 1 }
  lbuf[llen] = 0 as u8
  let keys: *i64 = sys_mmap(8 * 4) as *i64
  let vals: *i64 = sys_mmap(8 * 4) as *i64
  keys[0] = "kind" as *u8 as i64
  keys[1] = "id" as *u8 as i64
  keys[2] = "row" as *u8 as i64
  vals[0] = "queue-row" as *u8 as i64
  vals[1] = idb as i64
  vals[2] = lbuf as i64
  // EXACT, from the encoder's own arithmetic -- not a cap, not headroom.
  let enc: *u8 = sys_mmap(canon_encode_size(keys, vals, 3))
  let el: i64 = canon_encode(keys, vals, 3, enc)
  let kbuf: *u8 = sys_mmap(128)
  var ko: i64 = 0
  ko = ss_cat(kbuf, ko, "aq:" as *u8)
  ko = ss_cat(kbuf, ko, idb)
  kbuf[ko] = 0 as u8
  return ql_put(prefix, st, kbuf, enc, el)
123}
124
125func ql_load_queue(prefix: *u8, path: *u8, st: *i64) -> i64 {
  let szp: *i64 = sys_mmap(16) as *i64
  let b: *u8 = ss_readall(path, szp)
  let sz: i64 = szp[0]
  if sz <= 0 { return 0 - 1 }
  var i: i64 = 0
  while i < sz {
      var e: i64 = i
      var go: i64 = 1
      while go == 1 {
          if e >= sz { go = 0 } else {
              if b[e] == (10 as u8) { go = 0 } else { e = e + 1 }
          }
      }
      if ql_row(prefix, st, b, i, e) < 0 { return 0 - 2 }
      i = e + 1
  }
  return 0
143}
144
145// line-CID records for a sponsor-surface file (nx_log_load pattern)
146func ql_load_lines(prefix: *u8, path: *u8, tag: *u8, st: *i64) -> i64 {
  let szp: *i64 = sys_mmap(16) as *i64
  let b: *u8 = ss_readall(path, szp)
  let sz: i64 = szp[0]
  if sz <= 0 { return 0 }
  let keys: *i64 = sys_mmap(8 * 4) as *i64
  let vals: *i64 = sys_mmap(8 * 4) as *i64
  keys[0] = "kind" as *u8 as i64
  keys[1] = "log" as *u8 as i64
  keys[2] = "line" as *u8 as i64
  vals[0] = "sponsor-surface-line" as *u8 as i64
  vals[1] = tag as i64
  // Allocated ONCE and reused for every line -- sizing per line would allocate
  // inside the hot loop. No line can be longer than the file that contains it,
  // so sz is a proven-sufficient bound rather than a guessed one.
  let lbuf: *u8 = sys_mmap(sz + 1)
  lbuf[0] = 0 as u8
  vals[2] = lbuf as i64
  // canon_encode_size over an empty line yields the exact fixed cost of this
  // record's keys and values; the only quantity the encoder cannot know is the
  // longest line, and the file length bounds that exactly.
  let enc: *u8 = sys_mmap(canon_encode_size(keys, vals, 3) + sz)
  let cid: *u8 = sys_mmap(96)
  let kbuf: *u8 = sys_mmap(160)
  var i: i64 = 0
  while i < sz {
      var e: i64 = i
      var go: i64 = 1
      while go == 1 {
          if e >= sz { go = 0 } else {
              if b[e] == (10 as u8) { go = 0 } else { e = e + 1 }
          }
      }
      let llen: i64 = e - i
      var use: i64 = 1
      if llen <= 0 { use = 0 }
      // no length test: lbuf is sized from the file, so every line fits
      if use == 1 { if b[i] == (35 as u8) { use = 0 } }
      if use == 1 {
          // Sponsor-surface lines are counted SEPARATELY from queue rows. One
          // counter serving two populations cannot be checked against either:
          // a fixture of 3 queue rows read 131 because this pass shared st[5],
          // so the number could not confirm or refute anything about the queue.
          st[9] = st[9] + 1
          var t: i64 = 0
          while t < llen { lbuf[t] = b[i + t]; t = t + 1 }
          lbuf[llen] = 0 as u8
          vals[2] = lbuf as i64
          let el: i64 = canon_encode(keys, vals, 3, enc)
          cid_of(enc, el, cid)
          var ko: i64 = 0
          ko = ss_cat(kbuf, ko, tag)
          ko = ss_cat(kbuf, ko, ":" as *u8)
          ko = ss_cat(kbuf, ko, cid)
          kbuf[ko] = 0 as u8
          if ql_put(prefix, st, kbuf, enc, el) < 0 { return 0 - 2 }
      }
      i = e + 1
  }
  return 0
206}
207
208func main(argc: i64, argv: *i64) -> i64 {
  var prefix: *u8 = "knowledge/store/im-" as *u8
  var qpath: *u8 = "knowledge/registry/assignment_queue.tsv" as *u8
  if argc >= 2 { prefix = argv[1] as *u8 }
  if argc >= 3 { qpath = argv[2] as *u8 }
  let logfd: i64 = sys_openat_append("knowledge/status/infomgmt_store.log" as *u8, MODE_0644)
214
  let st: *i64 = sys_mmap(8 * 16) as *i64
  st[0] = ss_begin() as i64
  st[7] = ss_open(prefix) as i64
218
  if ql_load_queue(prefix, qpath, st) != 0 {
      ql_p("QUEUE-STORE FAILED rc-inner=" as *u8); ql_fn(1, st[6]); ql_p(" -- fail loud\n" as *u8)
      sys_exit(1)
  }
  // sponsor surfaces ride the same pass (real paths only; scratch gates
  // pass a scratch prefix so these dedup against it harmlessly)
  ql_load_lines(prefix, "knowledge/registry/sponsor_decisions.tsv" as *u8, "sd" as *u8, st)
  ql_load_lines(prefix, "knowledge/registry/planning_prefs.tsv" as *u8, "pp" as *u8, st)
  ql_load_lines(prefix, "knowledge/registry/callouts.tsv" as *u8, "co" as *u8, st)
  if st[6] != 0 {
      ql_p("QUEUE-STORE FAILED late rc=" as *u8); ql_fn(1, st[6]); ql_p("\n" as *u8)
      sys_exit(1)
  }
  if ql_flush(prefix, st) != 0 {
      ql_p("QUEUE-STORE final commit FAILED -- fail loud\n" as *u8)
      sys_exit(1)
  }
236
  // probe: the queue is QUERYABLE on the team's own substrate
  let h2: *i64 = ss_open(prefix)
  let pp2: *i64 = sys_mmap(16) as *i64
  let ll2: *i64 = sys_mmap(16) as *i64
  let probe: i64 = ss_hget(h2, "aq:X-Q-003" as *u8, pp2, ll2)
242
  var fdi: i64 = 0
  while fdi < 2 {
      var fd: i64 = 1
      if fdi == 1 { fd = logfd }
      if fd > 0 {
          ql_fp(fd, "QUEUE-STORE queue_rows=" as *u8); ql_fn(fd, st[5])
          ql_fp(fd, " sponsor_lines=" as *u8); ql_fn(fd, st[9])
          ql_fp(fd, " added=" as *u8); ql_fn(fd, st[3])
          ql_fp(fd, " dup_instore=" as *u8); ql_fn(fd, st[4])
          // Lines whose field 0 was too long to be a queue id. These are
          // expected (prose, banners) -- but they are REPORTED, because a line
          // this organ declined to ingest must never look like a line that was
          // never in the file.
          ql_fp(fd, " rejected_long_id=" as *u8); ql_fn(fd, st[8])
          ql_fp(fd, " segments=" as *u8); ql_fn(fd, st[2])
          ql_fp(fd, " probe_aq=" as *u8); ql_fn(fd, probe)
          ql_fp(fd, " epoch=" as *u8); ql_fn(fd, sys_now_realtime_sec())
          ql_fp(fd, " verdict=GREEN\n" as *u8)
      }
      fdi = fdi + 1
  }
  if logfd > 0 { sys_close(logfd) }
  sys_exit(0)
  return 0
267}
nx_queue_load.nx source

`nx_queue_load.nx` source
