`nx_feeder.nx` source

↩ module page · 605 lines · 24193 B
1// nx_feeder.nx -- AUTONOMOUS BACKLOG FEEDER, operational-gap half (ws=ws-intel8, 2026-07-20).
2// The INPUT side of the intake pipeline: until now a request only existed if a human or an
3// agent thought to file one. This organ makes the ECOSYSTEM NOTICE ITS OWN DROPPED THREADS
4// and file its own tickets, from signals it already records:
5//   (1) DROPPED WORKSTREAM  -- a ws_sync lane with a KICKOFF, no DONE, silent > threshold
6//   (2) EXPIRED LEASE       -- an nx_claims resource held past its TTL, never released
7// Each finding becomes ONE nx_pm_intake REQ carrying a STABLE MARKER, so a re-run never
8// re-files it (idempotent by construction), and each run is CAPPED so the queue can never
9// be flooded. NOTHING IS FABRICATED: every proposal cites its measured age/holder/checkpoint;
10// a source with no finding files nothing. Downstream: nx_triage routes it, /intake renders it.
11// Composes nx_ws_sync + nx_claims + nx_pm_intake; duplicates none of them.
12// F208 COORDINATION: the ATLAS-sourced feeder half (maturity-gap x momentum -> F-rows,
13// researcher+architect, nx_atlas_discover/recombine) is NOT claimed here -- different
14// sources, complementary; this is the operational-gap half owned by the coordination lane.
15// SCALE LAW: windowed reads + declared envelope (window_bytes/truncated/filed/skipped/capped).
16// license_tier: ORIGINAL   No hw writes (Rule 26).
17//   scan     <ws_jrnl> <claims_jrnl> <intake_jrnl>                  -> JSON findings (READ-ONLY)
18//   run      <ws_jrnl> <claims_jrnl> <intake_jrnl> <actor>          -> file them, JSON summary
19//   selftest <scratch-journal>                                      -> gate T1..T8 (pre-cleaned)
20import "nx_syscalls.nx"
21import "nx_gate_verdict.nx"
22const FD_MAGIC_4096: i64 = 4096
23const FD_MAGIC_2000: i64 = 2000
24const FD_MAGIC_1024: i64 = 1024
25const FD_MAGIC_1000000: i64 = 1000000
26
27const FD_WIN: i64 = 4194304
28const FD_OUT: i64 = 262144
29const FD_SOFT: i64 = 49152
30const FD_NOTE: i64 = 1024
31const FD_NAME_MAX: i64 = 120
32const FD_NOTE_MAX: i64 = 200
33// POLICY (residual: promote to config rows): a lane silent this long with no DONE is dropped.
34const FD_DROPPED_SEC: i64 = 172800
35const FD_MAXFILE: i64 = 5
36
37func fd_cat(d: *u8, o: i64, s: *u8) -> i64 { var p: i64=o; var i: i64=0; while s[i]!=(0 as u8){ d[p]=s[i]; p=p+1; i=i+1 } return p }
38func fd_catn(d: *u8, o: i64, v: i64) -> i64 {
  let t: *u8 = sys_mmap(28)
  var m: i64 = v
  var p: i64 = o
  if m < 0 { d[p]=45 as u8; p=p+1; m=0-m }
  var k: i64 = 0
  if m == 0 { t[0]=48 as u8; 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 { d[p]=t[k-1-i]; p=p+1; i=i+1 }
  return p
49}
50func fd_vlen(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} return n }
51func fd_read(path: *u8, buf: *u8, cap: i64) -> i64 {
  let fd: i64 = sys_openat_rd(path)
  if fd < 0 { return 0 }
  var n: i64 = 0
  var go: i64 = 1
  while go == 1 { let r: i64 = sys_read(fd, ((buf as i64)+n) as *u8, cap-n); if r <= 0 { go = 0 } else { n = n + r } if n >= cap { go = 0 } }
  sys_close(fd)
  return n
59}
60func fd_le(q: *u8, i: i64, n: i64) -> i64 {
  var le: i64 = i
  var s: i64 = 1
  while s==1 { if le>=n { s=0 } else { if q[le]==(10 as u8){s=0} else {le=le+1} } }
  return le
65}
66func fd_col(q: *u8, ls: i64, le: i64, c: i64, out: *i64) -> i64 {
  var col: i64 = 0
  var p: i64 = ls
  while col < c {
      var s: i64 = 1
      while s == 1 { if p >= le { return 0 } if q[p]==(9 as u8) { s = 0 } else { p = p+1 } }
      p = p + 1
      col = col + 1
  }
  var e: i64 = p
  var s2: i64 = 1
  while s2 == 1 { if e >= le { s2 = 0 } else { if q[e]==(9 as u8) { s2 = 0 } else { e = e+1 } } }
  out[0] = p
  out[1] = e
  return 1
81}
82func fd_lit_eq(q: *u8, s: i64, e: i64, lit: *u8) -> i64 {
  var i: i64 = 0
  while s+i < e { if lit[i]==(0 as u8) { return 0 } if q[s+i]!=lit[i] { return 0 } i=i+1 }
  if lit[i]!=(0 as u8) { return 0 }
  return 1
87}
88func fd_span_eq(q: *u8, s1: i64, e1: i64, s2: i64, e2: i64) -> i64 {
  if e1-s1 != e2-s2 { return 0 }
  var i: i64 = 0
  while s1+i < e1 { if q[s1+i]!=q[s2+i] { return 0 } i=i+1 }
  return 1
93}
94func fd_atoi(q: *u8, s: i64, e: i64) -> i64 {
  var v: i64 = 0
  var i: i64 = s
  while i < e { let c: i64 = q[i] as i64; if c>=48 { if c<=57 { v = v*10 + (c-48) } } i=i+1 }
  return v
99}
100// substring search of a NUL-terminated needle inside a buffer
101func fd_find(d: *u8, dn: i64, lit: *u8) -> i64 {
  let ll: i64 = fd_vlen(lit)
  if ll == 0 { return 0 }
  var i: i64 = 0
  while i + ll <= dn {
      var k: i64 = 0
      var ok: i64 = 1
      while k < ll { if d[i+k]!=lit[k] { ok=0; k=ll } else { k=k+1 } }
      if ok==1 { return 1 }
      i = i + 1
  }
  return 0
113}
114// append an nx_pm_intake REQ frame (contract: <ts>\t<VERB>\t<id>\t<actor>\t<text>)
115func fd_append_req(journal: *u8, id: i64, actor: *u8, note: *u8) -> i64 {
  let ln: *u8 = sys_mmap(FD_MAGIC_4096)
  var o: i64 = 0
  o = fd_catn(ln, o, id)
  ln[o]=9 as u8; o=o+1
  o = fd_cat(ln, o, "REQ" as *u8)
  ln[o]=9 as u8; o=o+1
  o = fd_catn(ln, o, id)
  ln[o]=9 as u8; o=o+1
  o = fd_cat(ln, o, actor)
  ln[o]=9 as u8; o=o+1
  var i: i64 = 0
  var used: i64 = 0
  while note[i]!=(0 as u8) {
      if used < FD_MAGIC_2000 {
          let c: i64 = note[i] as i64
          if c==9 { ln[o]=32 as u8; o=o+1 } else { if c==10 { ln[o]=32 as u8; o=o+1 } else { if c==13 { ln[o]=32 as u8; o=o+1 } else { ln[o]=c as u8; o=o+1 } } }
          used = used + 1
      }
      i = i + 1
  }
  ln[o]=10 as u8; o=o+1
  let fd: i64 = sys_openat_append(journal, 0x1a4)
  if fd < 0 { return -1 }
  sys_write(fd, ln, o)
  sys_close(fd)
  return 0
142}
143// does the intake journal already carry this REQ id? (collision-safe id minting)
144func fd_id_taken(q: *u8, n: i64, idv: i64) -> i64 {
  let ci: *i64 = sys_mmap(16) as *i64
  var i: i64 = 0
  while i < n {
      let le: i64 = fd_le(q,i,n)
      if fd_col(q,i,le,2,ci)==1 { let v: i64 = fd_atoi(q,ci[0],ci[1]); if v==idv { return 1 } }
      i = le + 1
  }
  return 0
153}
154
155// ---------- source 1: dropped workstreams (ws_sync contract) ----------
156func fd_ws_first_kick(q: *u8, upto: i64, ws_s: i64, ws_e: i64) -> i64 {
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  var i: i64 = 0
  while i < upto {
      let le: i64 = fd_le(q,i,upto)
      if fd_col(q,i,le,1,cv)==1 { if fd_lit_eq(q,cv[0],cv[1],"KICKOFF" as *u8)==1 {
          if fd_col(q,i,le,2,cw)==1 { if fd_span_eq(q,cw[0],cw[1],ws_s,ws_e)==1 { return 0 } }
      } }
      i = le + 1
  }
  return 1
168}
169func fd_ws_has(q: *u8, n: i64, verb: *u8, ws_s: i64, ws_e: i64) -> i64 {
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  var i: i64 = 0
  while i < n {
      let le: i64 = fd_le(q,i,n)
      if fd_col(q,i,le,1,cv)==1 { if fd_lit_eq(q,cv[0],cv[1],verb)==1 {
          if fd_col(q,i,le,2,cw)==1 { if fd_span_eq(q,cw[0],cw[1],ws_s,ws_e)==1 { return 1 } }
      } }
      i = le + 1
  }
  return 0
181}
182func fd_ws_last_ts(q: *u8, n: i64, ws_s: i64, ws_e: i64) -> i64 {
  let cw: *i64 = sys_mmap(16) as *i64
  let c0: *i64 = sys_mmap(16) as *i64
  var last: i64 = -1
  var i: i64 = 0
  while i < n {
      let le: i64 = fd_le(q,i,n)
      if fd_col(q,i,le,2,cw)==1 { if fd_span_eq(q,cw[0],cw[1],ws_s,ws_e)==1 {
          if fd_col(q,i,le,0,c0)==1 { last = fd_atoi(q,c0[0],c0[1]) }
      } }
      i = le + 1
  }
  return last
195}
196func fd_ws_last_note(q: *u8, n: i64, ws_s: i64, ws_e: i64, out: *i64) -> i64 {
  let cw: *i64 = sys_mmap(16) as *i64
  let cn: *i64 = sys_mmap(16) as *i64
  var ns: i64 = -1
  var ne: i64 = -1
  var i: i64 = 0
  while i < n {
      let le: i64 = fd_le(q,i,n)
      if fd_col(q,i,le,2,cw)==1 { if fd_span_eq(q,cw[0],cw[1],ws_s,ws_e)==1 {
          if fd_col(q,i,le,4,cn)==1 { ns=cn[0]; ne=cn[1] }
      } }
      i = le + 1
  }
  if ns >= 0 { out[0]=ns; out[1]=ne; return 1 }
  return 0
211}
212// is this lane DROPPED? (no DONE frame AND silent longer than the threshold)
213func fd_ws_dropped(q: *u8, n: i64, now: i64, ws_s: i64, ws_e: i64) -> i64 {
  if fd_ws_has(q,n,"DONE" as *u8,ws_s,ws_e)==1 { return 0 }
  let lt: i64 = fd_ws_last_ts(q,n,ws_s,ws_e)
  if lt < 0 { return 0 }
  if now - lt > FD_DROPPED_SEC { return 1 }
  return 0
219}
220
221// ---------- source 2: expired leases (nx_claims contract) ----------
222// state: -1 absent / 1 HELD / 2 EXPIRED / 3 FREE; out[0..1]=holder span, out[2]=ttl, out[3]=last
223func fd_cl_state(q: *u8, n: i64, now: i64, res_s: i64, res_e: i64, out: *i64) -> i64 {
  let cv: *i64 = sys_mmap(16) as *i64
  let cr: *i64 = sys_mmap(16) as *i64
  let ca: *i64 = sys_mmap(16) as *i64
  let ct: *i64 = sys_mmap(16) as *i64
  let c0: *i64 = sys_mmap(16) as *i64
  var mode: i64 = 0
  var ever: i64 = 0
  var hs: i64 = -1
  var he: i64 = -1
  var ttl: i64 = 0
  var last: i64 = 0
  var i: i64 = 0
  while i < n {
      let le: i64 = fd_le(q,i,n)
      if fd_col(q,i,le,2,cr)==1 { if fd_span_eq(q,cr[0],cr[1],res_s,res_e)==1 {
          if fd_col(q,i,le,1,cv)==1 { if fd_col(q,i,le,0,c0)==1 { if fd_col(q,i,le,3,ca)==1 {
              let ts: i64 = fd_atoi(q,c0[0],c0[1])
              if fd_lit_eq(q,cv[0],cv[1],"CLAIM" as *u8)==1 {
                  mode=1
                  ever=1
                  hs=ca[0]
                  he=ca[1]
                  last=ts
                  if fd_col(q,i,le,4,ct)==1 { ttl = fd_atoi(q,ct[0],ct[1]) }
              }
              if fd_lit_eq(q,cv[0],cv[1],"BEAT" as *u8)==1 { if mode==1 { if fd_span_eq(q,ca[0],ca[1],hs,he)==1 { last=ts } } }
              if fd_lit_eq(q,cv[0],cv[1],"RELEASE" as *u8)==1 { if mode==1 { if fd_span_eq(q,ca[0],ca[1],hs,he)==1 { mode=0 } } }
          } } }
      } }
      i = le + 1
  }
  if ever==0 { return -1 }
  if mode==0 { return 3 }
  out[0]=hs
  out[1]=he
  out[2]=ttl
  out[3]=last
  if now - last > ttl { return 2 }
  return 1
263}
264func fd_res_seen(q: *u8, upto: i64, r_s: i64, r_e: i64) -> i64 {
  let cr: *i64 = sys_mmap(16) as *i64
  var i: i64 = 0
  while i < upto {
      let le: i64 = fd_le(q,i,upto)
      if fd_col(q,i,le,2,cr)==1 { if fd_span_eq(q,cr[0],cr[1],r_s,r_e)==1 { return 1 } }
      i = le + 1
  }
  return 0
273}
274
275// ---------- proposal note builders (marker = the dedupe key) ----------
276func fd_note_dropped(nb: *u8, q: *u8, ws_s: i64, ws_e: i64, age_min: i64, note_s: i64, note_e: i64, hasnote: i64) -> i64 {
  var p: i64 = 0
  p = fd_cat(nb,p,"DROPPED WORKSTREAM (measured by nx_feeder): ws=" as *u8)
  var i: i64 = ws_s
  while i < ws_e { nb[p]=q[i]; p=p+1; i=i+1 }
  p = fd_cat(nb,p," has a KICKOFF but NO DONE frame and has been silent " as *u8)
  p = fd_catn(nb,p,age_min)
  p = fd_cat(nb,p," min. Decide: resume it, hand it off, or close it with a DONE frame. Last checkpoint: " as *u8)
  if hasnote==1 {
      var j: i64 = note_s
      var used: i64 = 0
      while j < note_e { if used < FD_NOTE_MAX { let c: i64 = q[j] as i64; if c==9 { nb[p]=32 as u8 } else { nb[p]=c as u8 } p=p+1; used=used+1 } j=j+1 }
  }
  p = fd_cat(nb,p," [FEEDER:dropped-ws:" as *u8)
  var k: i64 = ws_s
  while k < ws_e { nb[p]=q[k]; p=p+1; k=k+1 }
  p = fd_cat(nb,p,"]" as *u8)
  nb[p]=0 as u8
  return p
295}
296func fd_note_expired(nb: *u8, q: *u8, r_s: i64, r_e: i64, h_s: i64, h_e: i64, over_sec: i64, ttl: i64) -> i64 {
  var p: i64 = 0
  p = fd_cat(nb,p,"EXPIRED LEASE (measured by nx_feeder): resource=" as *u8)
  var i: i64 = r_s
  while i < r_e { nb[p]=q[i]; p=p+1; i=i+1 }
  p = fd_cat(nb,p," held by " as *u8)
  var j: i64 = h_s
  while j < h_e { nb[p]=q[j]; p=p+1; j=j+1 }
  p = fd_cat(nb,p," is " as *u8)
  p = fd_catn(nb,p,over_sec)
  p = fd_cat(nb,p,"s past its " as *u8)
  p = fd_catn(nb,p,ttl)
  p = fd_cat(nb,p,"s TTL with no RELEASE (crash-expiry). Reap it or re-claim it. [FEEDER:expired-claim:" as *u8)
  var k: i64 = r_s
  while k < r_e { nb[p]=q[k]; p=p+1; k=k+1 }
  p = fd_cat(nb,p,"]" as *u8)
  nb[p]=0 as u8
  return p
314}
315// marker-only string, for the dedupe probe
316func fd_marker_dropped(mb: *u8, q: *u8, ws_s: i64, ws_e: i64) -> i64 {
  var p: i64 = 0
  p = fd_cat(mb,p,"[FEEDER:dropped-ws:" as *u8)
  var i: i64 = ws_s
  while i < ws_e { mb[p]=q[i]; p=p+1; i=i+1 }
  p = fd_cat(mb,p,"]" as *u8)
  mb[p]=0 as u8
  return p
324}
325func fd_marker_expired(mb: *u8, q: *u8, r_s: i64, r_e: i64) -> i64 {
  var p: i64 = 0
  p = fd_cat(mb,p,"[FEEDER:expired-claim:" as *u8)
  var i: i64 = r_s
  while i < r_e { mb[p]=q[i]; p=p+1; i=i+1 }
  p = fd_cat(mb,p,"]" as *u8)
  mb[p]=0 as u8
  return p
333}
334
335// ---------- the sweep: mode 0 = count/report only, mode 1 = file ----------
336// cx[0]=filed cx[1]=skipped_dup cx[2]=capped cx[3]=findings cx[4]=next_id
337func fd_sweep(wq: *u8, wn: i64, cq: *u8, cn: i64, iq: *u8, iN: i64, ij: *u8, actor: *u8, now: i64, mode: i64, maxfile: i64, cx: *i64) -> i64 {
  cx[0]=0
  cx[1]=0
  cx[2]=0
  cx[3]=0
  let cv: *i64 = sys_mmap(16) as *i64
  let cw: *i64 = sys_mmap(16) as *i64
  let cr: *i64 = sys_mmap(16) as *i64
  let st: *i64 = sys_mmap(64) as *i64
  let no: *i64 = sys_mmap(16) as *i64
  let nb: *u8 = sys_mmap(FD_NOTE)
  let mb: *u8 = sys_mmap(512)
  var nextid: i64 = now
  // source 1: dropped workstreams
  var i: i64 = 0
  while i < wn {
      let le: i64 = fd_le(wq,i,wn)
      if fd_col(wq,i,le,1,cv)==1 { if fd_lit_eq(wq,cv[0],cv[1],"KICKOFF" as *u8)==1 {
          if fd_col(wq,i,le,2,cw)==1 { if fd_ws_first_kick(wq,i,cw[0],cw[1])==1 {
              if fd_ws_dropped(wq,wn,now,cw[0],cw[1])==1 {
                  cx[3] = cx[3] + 1
                  fd_marker_dropped(mb,wq,cw[0],cw[1])
                  if fd_find(iq,iN,mb)==1 { cx[1] = cx[1] + 1 } else {
                      if cx[0] >= maxfile { cx[2] = cx[2] + 1 } else {
                          if mode == 1 {
                              let lt: i64 = fd_ws_last_ts(wq,wn,cw[0],cw[1])
                              let agem: i64 = (now - lt) / 60
                              let hn: i64 = fd_ws_last_note(wq,wn,cw[0],cw[1],no)
                              fd_note_dropped(nb,wq,cw[0],cw[1],agem,no[0],no[1],hn)
                              var go: i64 = 1
                              while go == 1 { if fd_id_taken(iq,iN,nextid)==1 { nextid = nextid + 1 } else { go = 0 } }
                              let rc: i64 = fd_append_req(ij,nextid,actor,nb)
                              if rc==0 { cx[0] = cx[0] + 1; nextid = nextid + 1 }
                          } else { cx[0] = cx[0] + 1 }
                      }
                  }
              }
          } }
      } }
      i = le + 1
  }
  // source 2: expired leases
  var j: i64 = 0
  while j < cn {
      let je: i64 = fd_le(cq,j,cn)
      if fd_col(cq,j,je,2,cr)==1 { if fd_res_seen(cq,j,cr[0],cr[1])==0 {
          let s: i64 = fd_cl_state(cq,cn,now,cr[0],cr[1],st)
          if s == 2 {
              cx[3] = cx[3] + 1
              fd_marker_expired(mb,cq,cr[0],cr[1])
              if fd_find(iq,iN,mb)==1 { cx[1] = cx[1] + 1 } else {
                  if cx[0] >= maxfile { cx[2] = cx[2] + 1 } else {
                      if mode == 1 {
                          let over: i64 = (now - st[3]) - st[2]
                          fd_note_expired(nb,cq,cr[0],cr[1],st[0],st[1],over,st[2])
                          var go2: i64 = 1
                          while go2 == 1 { if fd_id_taken(iq,iN,nextid)==1 { nextid = nextid + 1 } else { go2 = 0 } }
                          let rc2: i64 = fd_append_req(ij,nextid,actor,nb)
                          if rc2==0 { cx[0] = cx[0] + 1; nextid = nextid + 1 }
                      } else { cx[0] = cx[0] + 1 }
                  }
              }
          }
      } }
      j = je + 1
  }
  cx[4] = nextid
  return cx[0]
405}
406func fd_key(d: *u8, o: i64, name: *u8) -> i64 {
  d[o]=34 as u8
  var p: i64 = o + 1
  p = fd_cat(d,p,name)
  d[p]=34 as u8
  p = p + 1
  d[p]=58 as u8
  p = p + 1
  return p
415}
416func fd_qlit(d: *u8, o: i64, s: *u8) -> i64 {
  var p: i64 = o
  d[p]=34 as u8; p=p+1
  p = fd_cat(d,p,s)
  d[p]=34 as u8; p=p+1
  return p
422}
423func fd_summary(d: *u8, modename: *u8, cx: *i64, wn: i64, cn: i64, iN: i64, trunc: i64, maxfile: i64) -> i64 {
  var p: i64 = 0
  d[p]=123 as u8; p=p+1
  p = fd_key(d,p,"v" as *u8)
  p = fd_catn(d,p,1)
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"tool" as *u8)
  p = fd_qlit(d,p,"nx_feeder" as *u8)
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"mode" as *u8)
  p = fd_qlit(d,p,modename)
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"findings" as *u8)
  p = fd_catn(d,p,cx[3])
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"filed" as *u8)
  p = fd_catn(d,p,cx[0])
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"skipped_already_filed" as *u8)
  p = fd_catn(d,p,cx[1])
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"capped" as *u8)
  p = fd_catn(d,p,cx[2])
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"max_per_run" as *u8)
  p = fd_catn(d,p,maxfile)
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"dropped_after_sec" as *u8)
  p = fd_catn(d,p,FD_DROPPED_SEC)
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"ws_bytes" as *u8)
  p = fd_catn(d,p,wn)
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"claims_bytes" as *u8)
  p = fd_catn(d,p,cn)
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"intake_bytes" as *u8)
  p = fd_catn(d,p,iN)
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"window_bytes" as *u8)
  p = fd_catn(d,p,FD_WIN)
  d[p]=44 as u8; p=p+1
  p = fd_key(d,p,"truncated" as *u8)
  p = fd_catn(d,p,trunc)
  d[p]=125 as u8; p=p+1
  d[p]=10 as u8; p=p+1
  return p
470}
471func fd_mkpath(dst: *u8, src: *u8, suf: *u8) -> i64 {
  var p: i64 = 0
  p = fd_cat(dst,p,src)
  p = fd_cat(dst,p,suf)
  dst[p]=0 as u8
  return p
477}
478func fd_wr(path: *u8, body: *u8) -> i64 {
  let fd: i64 = sys_openat_wr(path, 0x1a4)
  if fd < 0 { return -1 }
  sys_write(fd, body, fd_vlen(body))
  sys_close(fd)
  return 0
484}
485
486// ---------- selftest gate (T1..T8) ----------
487func fd_selftest(base: *u8) -> i64 {
  let ctr: *i64 = gv_ctr()
  gv_head("nx_feeder selftest -- autonomous gap-detection teeth (caller pre-cleans paths)" as *u8)
  let wq: *u8 = sys_mmap(FD_WIN)
  let cq: *u8 = sys_mmap(FD_WIN)
  let iq: *u8 = sys_mmap(FD_WIN)
  let cx: *i64 = sys_mmap(64) as *i64
  let wp: *u8 = sys_mmap(FD_MAGIC_1024)
  let cp: *u8 = sys_mmap(FD_MAGIC_1024)
  let ip: *u8 = sys_mmap(FD_MAGIC_1024)
  let rcap: i64 = FD_WIN - 8
  let now: i64 = FD_MAGIC_1000000
  fd_mkpath(wp, base, ".ws" as *u8)
  fd_mkpath(cp, base, ".claims" as *u8)
  fd_mkpath(ip, base, ".intake" as *u8)
  // T1 empty sources -> nothing found, nothing filed
  var wn: i64 = fd_read(wp,wq,rcap)
  var cn: i64 = fd_read(cp,cq,rcap)
  var iN: i64 = fd_read(ip,iq,rcap)
  fd_sweep(wq,wn,cq,cn,iq,iN,ip,"t" as *u8,now,0,FD_MAXFILE,cx)
  var ok1: i64 = 0
  if cx[3]==0 { if cx[0]==0 { ok1=1 } }
  gv_check("T1 empty sources find nothing fabricate nothing" as *u8, ok1, ctr)
  // seed ws: laneDrop (old kickoff, no done) - laneDone (old, DONE) - laneFresh (recent, no done)
  fd_wr(wp, "100\tKICKOFF\tlaneDrop\ts1\tstarted the drop lane\n200\tBEAT\tlaneDrop\ts1\tlast checkpoint here\n100\tKICKOFF\tlaneDone\ts2\tstarted\n300\tDONE\tlaneDone\ts2\tfinished cleanly\n999000\tKICKOFF\tlaneFresh\ts3\tjust started\n" as *u8)
  // seed claims: resExp (old claim, ttl 10, no release) - resHeld (fresh claim, ttl 100000)
  fd_wr(cp, "500\tCLAIM\tresExp\tsessA\t10\tclaimed long ago\n999500\tCLAIM\tresHeld\tsessB\t100000\tclaimed just now\n" as *u8)
  wn = fd_read(wp,wq,rcap)
  cn = fd_read(cp,cq,rcap)
  iN = fd_read(ip,iq,rcap)
  fd_sweep(wq,wn,cq,cn,iq,iN,ip,"t" as *u8,now,0,FD_MAXFILE,cx)
  // T2 exactly two findings: laneDrop + resExp
  var ok2: i64 = 0
  if cx[3]==2 { ok2=1 }
  gv_check("T2 finds exactly the dropped lane and expired lease" as *u8, ok2, ctr)
  // T3/T4/T5 neg-controls proven by the run's written frames
  fd_sweep(wq,wn,cq,cn,iq,iN,ip,"nx_feeder" as *u8,now,1,FD_MAXFILE,cx)
  iN = fd_read(ip,iq,rcap)
  var ok3: i64 = 0
  if fd_find(iq,iN,"[FEEDER:dropped-ws:laneDrop]" as *u8)==1 { if fd_find(iq,iN,"[FEEDER:dropped-ws:laneDone]" as *u8)==0 { ok3=1 } }
  gv_check("T3 neg-control DONE lane never proposed" as *u8, ok3, ctr)
  var ok4: i64 = 0
  if fd_find(iq,iN,"[FEEDER:dropped-ws:laneFresh]" as *u8)==0 { ok4=1 }
  gv_check("T4 neg-control fresh lane never proposed" as *u8, ok4, ctr)
  var ok5: i64 = 0
  if fd_find(iq,iN,"[FEEDER:expired-claim:resExp]" as *u8)==1 { if fd_find(iq,iN,"[FEEDER:expired-claim:resHeld]" as *u8)==0 { ok5=1 } }
  gv_check("T5 expired lease proposed held lease never" as *u8, ok5, ctr)
  // T6 IDEMPOTENCE: a second run files nothing new, marks both as already-filed
  fd_sweep(wq,wn,cq,cn,iq,iN,ip,"nx_feeder" as *u8,now,1,FD_MAXFILE,cx)
  var ok6: i64 = 0
  if cx[0]==0 { if cx[1]==2 { ok6=1 } }
  gv_check("T6 neg-control rerun files nothing already-filed 2" as *u8, ok6, ctr)
  // T7 per-run CAP honored (maxfile 1 against a clean intake)
  fd_wr(ip, "# cleared for the cap tooth\n" as *u8)
  iN = fd_read(ip,iq,rcap)
  fd_sweep(wq,wn,cq,cn,iq,iN,ip,"nx_feeder" as *u8,now,1,1,cx)
  var ok7: i64 = 0
  if cx[0]==1 { if cx[2]==1 { ok7=1 } }
  gv_check("T7 per-run cap honored and reported" as *u8, ok7, ctr)
  // T8 frames are readable under the intake contract (REQ verb + id in cols)
  iN = fd_read(ip,iq,rcap)
  let cvv: *i64 = sys_mmap(16) as *i64
  var reqs: i64 = 0
  var z: i64 = 0
  while z < iN {
      let ze: i64 = fd_le(iq,z,iN)
      if fd_col(iq,z,ze,1,cvv)==1 { if fd_lit_eq(iq,cvv[0],cvv[1],"REQ" as *u8)==1 { reqs = reqs + 1 } }
      z = ze + 1
  }
  var ok8: i64 = 0
  if reqs==1 { ok8=1 }
  gv_check("T8 filed frames parse under the intake REQ contract" as *u8, ok8, ctr)
  let rc: i64 = gv_verdict("FEEDER-GATE" as *u8, ctr, "autonomous gap-detection teeth green" as *u8)
  return rc
561}
562
563func main(argc: i64, argv: *i64) -> i64 {
  if argc < 3 { gv_puts("usage: nx_feeder {scan <ws_jrnl> <claims_jrnl> <intake_jrnl> | run <ws_jrnl> <claims_jrnl> <intake_jrnl> <actor> | selftest <base>}\n" as *u8); sys_exit(2); return 2 }
  let verb: *u8 = argv[1] as *u8
  let vl: i64 = fd_vlen(verb)
  if fd_lit_eq(verb,0,vl,"selftest" as *u8)==1 { let rc: i64 = fd_selftest(argv[2] as *u8); sys_exit(rc); return rc }
  if argc < 5 { gv_puts("scan|run need <ws_jrnl> <claims_jrnl> <intake_jrnl>\n" as *u8); sys_exit(2); return 2 }
  let wp: *u8 = argv[2] as *u8
  let cp: *u8 = argv[3] as *u8
  let ip: *u8 = argv[4] as *u8
  let wq: *u8 = sys_mmap(FD_WIN)
  let cq: *u8 = sys_mmap(FD_WIN)
  let iq: *u8 = sys_mmap(FD_WIN)
  let rcap: i64 = FD_WIN - 8
  let wn: i64 = fd_read(wp,wq,rcap)
  let cn: i64 = fd_read(cp,cq,rcap)
  let iN: i64 = fd_read(ip,iq,rcap)
  var trunc: i64 = 0
  if wn >= rcap { trunc = 1 }
  if cn >= rcap { trunc = 1 }
  if iN >= rcap { trunc = 1 }
  let now: i64 = sys_now_realtime_sec()
  let cx: *i64 = sys_mmap(64) as *i64
  let d: *u8 = sys_mmap(FD_OUT)
  if fd_lit_eq(verb,0,vl,"scan" as *u8)==1 {
      fd_sweep(wq,wn,cq,cn,iq,iN,ip,"scan" as *u8,now,0,FD_MAXFILE,cx)
      let dl: i64 = fd_summary(d,"scan" as *u8,cx,wn,cn,iN,trunc,FD_MAXFILE)
      sys_write(1,d,dl)
      sys_exit(0)
      return 0
  }
  if fd_lit_eq(verb,0,vl,"run" as *u8)==1 {
      if argc < 6 { gv_puts("run needs <ws_jrnl> <claims_jrnl> <intake_jrnl> <actor>\n" as *u8); sys_exit(2); return 2 }
      let actor: *u8 = argv[5] as *u8
      fd_sweep(wq,wn,cq,cn,iq,iN,ip,actor,now,1,FD_MAXFILE,cx)
      let dl: i64 = fd_summary(d,"run" as *u8,cx,wn,cn,iN,trunc,FD_MAXFILE)
      sys_write(1,d,dl)
      sys_exit(0)
      return 0
  }
  gv_puts("unknown verb\n" as *u8)
  sys_exit(2)
  return 2
605}
nx_feeder.nx source

`nx_feeder.nx` source
