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

↩ module page · 216 lines · 11698 B
1// nx_research_serve.nx -- THE RESEARCH REQUEST QUEUE + ADEQUACY FEEDBACK GATE. Lets the autonomous builder and
2// the researcher partner gracefully + loosely (async, via artifacts, integer-only / no floats):
3//   - Auto-builder ENQUEUES by appending to knowledge/index/research_queue.txt:  <id> PENDING <topic terms...>
4//   - This organ DRAINS the queue: each PENDING request is queried across EVERY index (via the search-sources
5//     manifest = the same loose federation) and JUDGED for adequacy (distinct docs covering the topic >=
6//     SV_MIN_HITS), then routed:
7//        ADEQUATE   -> status SERVED      + research_served.log  (builder may now retrieve via federation)
8//        INADEQUATE -> status INADEQUATE  + research_gaps.txt    (THE FEEDBACK FLAG: under-covered topic -> shown
9//                                                                 to operator + Claude -> improve researcher / fetch deeper)
10//   - Queue rewritten with updated statuses (history kept; idempotent: only PENDING processed).
11// Closes the partner loop: builder asks -> researcher serves or flags the gap -> we grow the researcher -> the
12// auto-builder is unblocked -> the team grows toward autonomous S-class. expect_exit: 0  license_tier: ORIGINAL
13import "nx_search_inverted_persist.nx"
14import "nx_itoa_lib.nx"   // shared MSB-first emitter (zero-alloc)
15const SV_MAGIC_1024: i64 = 1024
16const SV_MAGIC_2048: i64 = 2048
17
18const SV_MIN_HITS: i64 = 3        // distinct docs to call research ADEQUATE (data-driven; config later)
19const SV_TERMS:    i64 = 8
20const SV_ROWIDS:   i64 = 4096
21const SV_IDX:      i64 = 8192
22const SV_HITS:     i64 = 65536
23
24func sv_puts(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
25// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
26// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
27// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
28// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
29func sv_num(v: i64) -> i64 { nxi_out(v); return 0 }
30func sv_strlen(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} return n }
31func sv_cat(dst: *u8, off: i64, s: *u8) -> i64 { var i: i64=0; while s[i]!=(0 as u8){dst[off+i]=s[i];i=i+1} return off+i }
32func sv_catbuf(dst: *u8, off: i64, src: *u8, n: i64) -> i64 { var i: i64=0; while i<n { dst[off+i]=src[i]; i=i+1 } return off+n }
33func sv_catnum(dst: *u8, off: i64, v: i64) -> i64 {
  if v == 0 { dst[off]=48 as u8; return off+1 }
  let tmp: *u8=sys_mmap(28); var m: i64=v; var k: i64=0
  while m>0 { tmp[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
  var i: i64=0; while i<k { dst[off+i]=tmp[k-1-i]; i=i+1 }
  return off+k
39}
40func sv_scan(buf: *u8, i: i64, end: i64) -> i64 {
  var p: i64=i; var stop: i64=0
  while stop==0 { if p>=end { stop=1 } else { if buf[p]==(32 as u8) { stop=1 } else { if buf[p]==(10 as u8) { stop=1 } else { p=p+1 } } } }
  return p
44}
45func sv_line_end(buf: *u8, i: i64, end: i64) -> i64 {
  var p: i64=i; var stop: i64=0
  while stop==0 { if p>=end { stop=1 } else { if buf[p]==(10 as u8) { stop=1 } else { p=p+1 } } }
  return p
49}
50func sv_load_docmap(path: *u8, paths_out: *i64, cap: i64) -> i64 {
  let lenbox: *i64 = sys_mmap(16) as *i64
  let buf: *u8 = sys_read_file(path, lenbox)
  if buf == 0 as *u8 { return 0 - 1 }
  let total: i64 = lenbox[0]
  var nd: i64 = 0; var start: i64 = 0; var i: i64 = 0
  while i < total {
      if buf[i] == (10 as u8) { buf[i]=0 as u8; if i>start { if nd<cap { paths_out[nd]=(buf as i64)+start; nd=nd+1 } } start=i+1 }
      i = i + 1
  }
  return nd
61}
62
63// load every index in the manifest (once) into the caller's arrays. Returns count.
64func sv_load_indexes(gidx: *i64, gpaths: *i64, gndocs: *i64) -> i64 {
  let mbox: *i64 = sys_mmap(16) as *i64
  let man: *u8 = sys_read_file("knowledge/index/search_sources.txt" as *u8, mbox)
  if man == 0 as *u8 { return 0 }
  let mlen: i64 = mbox[0]
  var n: i64 = 0; var i: i64 = 0
  while i < mlen {
      let e1: i64 = sv_scan(man, i, mlen)
      var j: i64 = e1; if j<mlen { if man[j]==(32 as u8) { man[j]=0 as u8; j=j+1 } }
      let idx_ptr: i64 = (man as i64) + j
      let e2: i64 = sv_scan(man, j, mlen)
      var k: i64 = e2; if k<mlen { if man[k]==(32 as u8) { man[k]=0 as u8; k=k+1 } }
      let dmap_ptr: i64 = (man as i64) + k
      let e3: i64 = sv_scan(man, k, mlen); if e3<mlen { if man[e3]==(32 as u8) { man[e3]=0 as u8 } }
      var e: i64 = sv_line_end(man, e3, mlen); if e<mlen { man[e]=0 as u8; e=e+1 }
      let idx: *NxInvIndex = nx_inv_load(idx_ptr as *u8)
      if idx != 0 as *NxInvIndex {
          let paths: *i64 = sys_mmap(8*SV_ROWIDS) as *i64
          let nd: i64 = sv_load_docmap(dmap_ptr as *u8, paths, SV_ROWIDS)
          if nd > 0 { if n < SV_IDX { gidx[n]=idx as i64; gpaths[n]=paths as i64; gndocs[n]=nd; n=n+1 } }
      }
      i = e
  }
  return n
88}
89
90// adequacy = distinct docs across all indexes matching ALL topic terms (conjunction). ANY-term matching is too
91// loose -- a nonsense topic with one common word ("widget") would falsely pass; requiring every term in the same
92// doc is what tells a real, covered topic from a stray match.
93func sv_hits_for(terms: *i64, nterms: i64, nidx: i64, gidx: *i64, gndocs: *i64, ghits: *u8, gcount: *u8) -> i64 {
  if nterms <= 0 { return 0 }
  let res: *NxInvQueryResult = sys_mmap(64) as *NxInvQueryResult
  let rowids: *i64 = sys_mmap(8*SV_ROWIDS) as *i64
  var total: i64 = 0
  var s: i64 = 0
  while s < nidx {
      let idx: *NxInvIndex = gidx[s] as *NxInvIndex
      let nd: i64 = gndocs[s]
      var z: i64 = 0
      while z < nd { gcount[z]=0 as u8; z=z+1 }              // # distinct terms matching each doc
      var t: i64 = 0
      while t < nterms {
          var z2: i64 = 0
          while z2 < nd { ghits[z2]=0 as u8; z2=z2+1 }        // per-term doc marker (dedup within term)
          let term: *u8 = terms[t] as *u8
          nx_inv_query_term(idx, term, sv_strlen(term), rowids, SV_ROWIDS, res)
          var ri: i64 = 0
          while ri < res.n_rowids_filled {
              let rid: i64 = rowids[ri]
              if rid >= 0 { if rid < nd { ghits[rid]=1 as u8 } }
              ri = ri + 1
          }
          var d2: i64 = 0
          while d2 < nd { if ghits[d2]==(1 as u8) { gcount[d2]=(gcount[d2] as i64 + 1) as u8 } d2=d2+1 }
          t = t + 1
      }
      var d: i64 = 0
      while d < nd { if gcount[d] as i64 == nterms { total=total+1 } d=d+1 }
      s = s + 1
  }
  return total
125}
126
127func main() -> i64 {
  sv_puts("=== nx_research_serve: drain research queue, judge adequacy, flag gaps (no floats) ===\n" as *u8)
  let gidx:   *i64 = sys_mmap(8*SV_IDX) as *i64
  let gpaths: *i64 = sys_mmap(8*SV_IDX) as *i64
  let gndocs: *i64 = sys_mmap(8*SV_IDX) as *i64
  let ghits:  *u8  = sys_mmap(SV_HITS)
  let gcount: *u8  = sys_mmap(SV_HITS)
  let nidx: i64 = sv_load_indexes(gidx, gpaths, gndocs)
  if nidx <= 0 { sv_puts(" NO indexes (run nx_search_register first)\n" as *u8); sys_exit(1); return 1 }
  sv_puts(" loaded indexes="); sv_num(nidx); sv_puts("\n" as *u8)
137
  let qbox: *i64 = sys_mmap(16) as *i64
  let q: *u8 = sys_read_file("knowledge/index/research_queue.txt" as *u8, qbox)
  if q == 0 as *u8 { sv_puts(" queue empty (enqueue: append '<id> PENDING <topic>' to research_queue.txt)\n" as *u8); sys_exit(0); return 0 }
  let qlen: i64 = qbox[0]
142
  let nq_fd: i64 = sys_openat_wr("knowledge/index/research_queue.next" as *u8, 0x1a4)
  let served_fd: i64 = sys_openat_append("knowledge/index/research_served.log" as *u8, 0x1a4)
  let gaps_fd: i64 = sys_openat_append("knowledge/index/research_gaps.txt" as *u8, 0x1a4)
146
  var served: i64=0; var inadequate: i64=0; var kept: i64=0
  var i: i64 = 0
  while i < qlen {
      let lstart: i64 = i
      let le: i64 = sv_line_end(q, i, qlen)
      if le > lstart {
          let e1: i64 = sv_scan(q, lstart, le)
          let id_ptr: i64 = (q as i64) + lstart; let id_len: i64 = e1 - lstart
          var j: i64 = e1; if j<le { if q[j]==(32 as u8) { j=j+1 } }
          let e2: i64 = sv_scan(q, j, le)
          let st_len: i64 = e2 - j
          var p: i64 = e2; if p<le { if q[p]==(32 as u8) { p=p+1 } }
          let topic_ptr: i64 = (q as i64) + p; let topic_len: i64 = le - p
160
          var pending: i64 = 0
          if st_len == 7 { if q[j]==(80 as u8) { pending=1 } }
          var ns_ptr: i64 = (q as i64) + j
          var ns_len: i64 = st_len
          if pending == 1 {
              let tbuf: *u8 = sys_mmap(topic_len + 8)
              var ci: i64 = 0
              while ci < topic_len { tbuf[ci]=q[p+ci]; ci=ci+1 }
              tbuf[topic_len]=0 as u8
              let terms: *i64 = sys_mmap(8*SV_TERMS) as *i64
              var nterms: i64 = 0
              var ti: i64 = 0
              var instr: i64 = 0
              while ti < topic_len {
                  if tbuf[ti]==(32 as u8) { tbuf[ti]=0 as u8; instr=0 }
                  else { if instr==0 { if nterms<SV_TERMS { terms[nterms]=((tbuf as i64)+ti); nterms=nterms+1 } instr=1 } }
                  ti=ti+1
              }
              let hits: i64 = sv_hits_for(terms, nterms, nidx, gidx, gndocs, ghits, gcount)
              if hits >= SV_MIN_HITS {
                  ns_ptr = "SERVED" as *u8 as i64; ns_len = 6; served=served+1
                  let sl: *u8=sys_mmap(SV_MAGIC_1024); var sp: i64=0
                  sp=sv_catbuf(sl,sp,id_ptr as *u8,id_len); sl[sp]=32 as u8; sp=sp+1
                  sp=sv_cat(sl,sp,"SERVED hits=" as *u8); sp=sv_catnum(sl,sp,hits); sl[sp]=32 as u8; sp=sp+1
                  sp=sv_catbuf(sl,sp,topic_ptr as *u8,topic_len); sl[sp]=10 as u8; sp=sp+1
                  if served_fd>=0 { sys_write(served_fd, sl, sp) }
              } else {
                  ns_ptr = "INADEQUATE" as *u8 as i64; ns_len = 10; inadequate=inadequate+1
                  let gl: *u8=sys_mmap(SV_MAGIC_1024); var gp: i64=0
                  gp=sv_cat(gl,gp,"GAP topic=" as *u8); gp=sv_catbuf(gl,gp,topic_ptr as *u8,topic_len)
                  gp=sv_cat(gl,gp," hits=" as *u8); gp=sv_catnum(gl,gp,hits)
                  gp=sv_cat(gl,gp," need>=" as *u8); gp=sv_catnum(gl,gp,SV_MIN_HITS)
                  gp=sv_cat(gl,gp," FLAG: improve researcher / fetch deeper for this topic\n" as *u8)
                  if gaps_fd>=0 { sys_write(gaps_fd, gl, gp) }
              }
          } else { kept = kept + 1 }
197
          let nl: *u8=sys_mmap(SV_MAGIC_2048); var np: i64=0
          np=sv_catbuf(nl,np,id_ptr as *u8,id_len); nl[np]=32 as u8; np=np+1
          np=sv_catbuf(nl,np,ns_ptr as *u8,ns_len); nl[np]=32 as u8; np=np+1
          np=sv_catbuf(nl,np,topic_ptr as *u8,topic_len); nl[np]=10 as u8; np=np+1
          if nq_fd>=0 { sys_write(nq_fd, nl, np) }
      }
      i = le; if i < qlen { i = i + 1 }
  }
  if nq_fd>=0 { sys_close(nq_fd) }
  if served_fd>=0 { sys_close(served_fd) }
  if gaps_fd>=0 { sys_close(gaps_fd) }
  sys_renameat("knowledge/index/research_queue.next" as *u8, "knowledge/index/research_queue.txt" as *u8)
210
  sv_puts(" drained: served="); sv_num(served); sv_puts(" inadequate_flagged="); sv_num(inadequate)
  sv_puts(" already_done="); sv_num(kept); sv_puts("\n" as *u8)
  sv_puts(" SERVED->research_served.log ; GAPS->research_gaps.txt (feedback flag to operator+Claude)\n" as *u8)
  sv_puts(" SERVE-OK\n" as *u8)
  sys_exit(0); return 0
216}
nx_research_serve.nx source

`nx_research_serve.nx` source
