code wiki / _hdl_build / nx_research_grow.nx
nx_research_grow.nx source
↩ module page · 318 lines · 16427 B
1// nx_research_grow.nx -- THE SELF-GROWING RESEARCHER INDEX (R4 self-grow loop) WITH CONTENT-DEDUP + CROSS-TOPIC
2// FLAGGING. Each run indexes the DELTA (fetched docs not yet SEEN) into a NEW append-only SHARD, but FIRST
3// fingerprints each doc's CONTENT (FNV-1a 64-bit). If that content fingerprint is already registered, the doc is
4// NOT re-indexed (no duplicate as coverage grows) -- instead the registry records the new (topic,path) for that
5// fingerprint, and if the new topic differs from the canonical topic we EMIT A CROSS-TOPIC FLAG: "this
6// information serves two research topics, found in the same place." So the corpus compounds on every use, never
7// duplicates content, never loses data, and the structure intelligently links topics that share sources.
8// Bounded batch per run = OOM-safe (no sys_munmap). Reuses the FIXED nx_inv build + nx_dirent walk.
9// Topic = the filename prefix before the first '_' (perma_/sem_/food_/ng_/...). expect_exit: 0 license_tier: ORIGINAL
10import "nx_search_inverted_persist.nx"
11import "nx_itoa_lib.nx" // shared MSB-first emitter (zero-alloc)
12import "nx_dirent.nx"
13
14const RG_DIR_BUF: i64 = 65536
15const RG_PATH_CAP: i64 = 4096
16const RG_BATCH_MAX: i64 = 50 // bounded files-read per run -> OOM-safe; corpus compounds across runs
17const RG_REG_CAP: i64 = 65536 // max distinct content fingerprints tracked in memory
18const RG_TOP_BUF: i64 = 4194304 // canonical-topic string arena
19
20func rg_puts(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
21// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
22// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
23// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
24// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
25func rg_num(v: i64) -> i64 { nxi_out(v); return 0 }
26func rg_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 }
27func rg_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 }
28func rg_catnum(dst: *u8, off: i64, v: i64) -> i64 {
29 if v == 0 { dst[off]=48 as u8; return off+1 }
30 let tmp: *u8=sys_mmap(28); var m: i64=v; if m<0 { dst[off]=45 as u8; off=off+1; m=0-m } var k: i64=0
31 while m>0 { tmp[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
32 var i: i64=0; while i<k { dst[off+i]=tmp[k-1-i]; i=i+1 }
33 return off+k
34}
35
36// indexable: .raw .md .txt
37func rg_indexable(name: *u8, n: i64) -> i64 {
38 if n > 3 { if name[n-3]==(46 as u8) { if name[n-2]==(109 as u8) { if name[n-1]==(100 as u8) { return 1 } } } }
39 if n > 4 { if name[n-4]==(46 as u8) { if name[n-3]==(116 as u8) { if name[n-2]==(120 as u8) { if name[n-1]==(116 as u8) { return 1 } } } } }
40 if n > 4 { if name[n-4]==(46 as u8) { if name[n-3]==(114 as u8) { if name[n-2]==(97 as u8) { if name[n-1]==(119 as u8) { return 1 } } } } }
41 return 0
42}
43
44// topic = filename prefix before first '_' (or whole name). Copies into out, returns len.
45func rg_topic(name: *u8, nlen: i64, out: *u8) -> i64 {
46 var i: i64=0; var stop: i64=0
47 while stop==0 {
48 if i >= nlen { stop=1 }
49 else {
50 if name[i]==(95 as u8) { stop=1 }
51 else { out[i]=name[i]; i=i+1 }
52 }
53 }
54 return i
55}
56
57func rg_streq(a: *u8, alen: i64, b: *u8, blen: i64) -> i64 {
58 if alen != blen { return 0 }
59 var i: i64=0
60 while i < alen { if a[i] != b[i] { return 0 } i=i+1 }
61 return 1
62}
63
64// is `path` (len plen) present as a full line in the manifest buffer? (SEEN-path dedup)
65func rg_seen(man: *u8, mlen: i64, path: *u8, plen: i64) -> i64 {
66 var i: i64=0; var lstart: i64=0; var found: i64=0
67 while i < mlen {
68 if man[i]==(10 as u8) {
69 if i - lstart == plen {
70 var j: i64=0; var same: i64=1
71 while j < plen { if man[lstart+j]!=path[j] { same=0 } j=j+1 }
72 if same==1 { found=1 }
73 }
74 lstart=i+1
75 }
76 i=i+1
77 }
78 return found
79}
80
81// scan a buffer field: advance from i to next space/newline/end
82func rg_field_end(buf: *u8, i: i64, end: i64) -> i64 {
83 var p: i64=i; var stop: i64=0
84 while stop==0 {
85 if p >= end { stop=1 }
86 else {
87 if buf[p]==(32 as u8) { stop=1 }
88 else { if buf[p]==(10 as u8) { stop=1 } else { p=p+1 } }
89 }
90 }
91 return p
92}
93func rg_parse_dec(buf: *u8, start: i64, fend: i64) -> i64 {
94 var v: i64=0; var p: i64=start
95 while p < fend { let c: i64=buf[p] as i64; if c>=48 { if c<=57 { v=v*10+(c-48) } } p=p+1 }
96 return v
97}
98
99// flatten one doc to a single corpus row + terminating \n; return 1 if non-empty written
100func rg_emit(corpus_fd: i64, buf: *u8, len: i64) -> i64 {
101 if len <= 0 { return 0 }
102 let row: *u8=sys_mmap(len+1); var nonspace: i64=0; var i: i64=0
103 while i < len {
104 var c: i64=buf[i] as i64
105 if c==10 { c=32 }
106 if c==13 { c=32 }
107 if c==9 { c=32 }
108 if c!=32 { nonspace=1 }
109 row[i]=c as u8; i=i+1
110 }
111 if nonspace==0 { return 0 }
112 row[len]=10 as u8
113 var off: i64=0
114 while off < len+1 { let w: i64=sys_write(corpus_fd, ((row as i64)+off) as *u8, len+1-off); if w<=0 { return 0-1 } off=off+w }
115 return 1
116}
117
118// content fingerprint: FNV-1a 64-bit over the FULL document bytes (NOT nx_inv_hash_bytes_lower, which caps at 64
119// bytes = a TOKEN hash -> every Wikipedia page shares the boilerplate prefix -> false dups -> data loss). This
120// hashes ALL len bytes so distinct documents get distinct fingerprints; only byte-identical content collides.
121func rg_fp(buf: *u8, len: i64) -> i64 {
122 var h: i64 = NX_INV_FNV1A_OFFSET_BASIS
123 var i: i64 = 0
124 while i < len {
125 let c: i64 = buf[i] as i64
126 h = h ^ c
127 h = h * NX_INV_FNV1A_PRIME
128 i = i + 1
129 }
130 if h < 0 { h = 0 - h }
131 return h
132}
133
134func main() -> i64 {
135 rg_puts("=== nx_research_grow: self-growing index + content-dedup + cross-topic flagging ===\n" as *u8)
136 sys_mkdir("knowledge/index" as *u8, 0x1ed)
137 sys_mkdir("knowledge/index/shards" as *u8, 0x1ed)
138
139 let man_path: *u8 = "knowledge/index/research_manifest.txt" as *u8
140 let led_path: *u8 = "knowledge/index/research_growth.log" as *u8
141 let reg_path: *u8 = "knowledge/index/content_registry.txt" as *u8
142 let flag_path: *u8 = "knowledge/index/cross_topic_flags.log" as *u8
143
144 // ---- load manifest (SEEN paths) ----
145 let mbox: *i64 = sys_mmap(16) as *i64
146 var man: *u8 = sys_read_file(man_path, mbox)
147 var mlen: i64 = 0
148 if man != 0 as *u8 { mlen = mbox[0] }
149 if man == 0 as *u8 { man = sys_mmap(16) }
150
151 // ---- load content registry -> unique fingerprints + canonical topic per fp ----
152 let fps: *i64 = sys_mmap(8*RG_REG_CAP) as *i64
153 let toff: *i64 = sys_mmap(8*RG_REG_CAP) as *i64
154 let tlen: *i64 = sys_mmap(8*RG_REG_CAP) as *i64
155 let topbuf: *u8 = sys_mmap(RG_TOP_BUF)
156 var tpos: i64 = 0
157 var nreg: i64 = 0
158 let rbox: *i64 = sys_mmap(16) as *i64
159 let reg: *u8 = sys_read_file(reg_path, rbox)
160 if reg != 0 as *u8 {
161 let rlen: i64 = rbox[0]
162 var i: i64 = 0
163 while i < rlen {
164 let f1: i64 = rg_field_end(reg, i, rlen) // fp field [i,f1)
165 let fp: i64 = rg_parse_dec(reg, i, f1)
166 var j: i64 = f1
167 if j < rlen { if reg[j]==(32 as u8) { j=j+1 } }
168 let f2: i64 = rg_field_end(reg, j, rlen) // topic field [j,f2)
169 let tl: i64 = f2 - j
170 // advance to next line
171 var e: i64 = f2
172 var st: i64 = 0
173 while st==0 { if e>=rlen { st=1 } else { if reg[e]==(10 as u8) { st=1 } else { e=e+1 } } }
174 if e < rlen { e=e+1 }
175 // register fp first-occurrence only (canonical)
176 var exists: i64 = 0
177 var k: i64 = 0
178 while k < nreg { if fps[k]==fp { exists=1 } k=k+1 }
179 if exists==0 { if nreg < RG_REG_CAP { fps[nreg]=fp; toff[nreg]=tpos; tlen[nreg]=tl; tpos=rg_catbuf(topbuf,tpos,((reg as i64)+j) as *u8,tl); nreg=nreg+1 } }
180 i = e
181 }
182 }
183 let unique_before: i64 = nreg
184
185 // ---- shard tag = ledger line count ----
186 let lbox: *i64 = sys_mmap(16) as *i64
187 let led: *u8 = sys_read_file(led_path, lbox)
188 var shard_n: i64=0
189 if led != 0 as *u8 { var li: i64=0; while li < lbox[0] { if led[li]==(10 as u8){shard_n=shard_n+1} li=li+1 } }
190
191 // ---- shard file paths ----
192 let corpus_path: *u8=sys_mmap(RG_PATH_CAP)
193 var cp: i64=rg_cat(corpus_path,0,"knowledge/index/shards/research_" as *u8); cp=rg_catnum(corpus_path,cp,shard_n); cp=rg_cat(corpus_path,cp,"_corpus.txt" as *u8); corpus_path[cp]=0 as u8
194 let idx_path: *u8=sys_mmap(RG_PATH_CAP)
195 var ipx: i64=rg_cat(idx_path,0,"knowledge/index/shards/research_" as *u8); ipx=rg_catnum(idx_path,ipx,shard_n); ipx=rg_cat(idx_path,ipx,".idx" as *u8); idx_path[ipx]=0 as u8
196 let docmap_path: *u8=sys_mmap(RG_PATH_CAP)
197 var dpx: i64=rg_cat(docmap_path,0,"knowledge/index/shards/research_" as *u8); dpx=rg_catnum(docmap_path,dpx,shard_n); dpx=rg_cat(docmap_path,dpx,".docmap" as *u8); docmap_path[dpx]=0 as u8
198
199 let corpus_fd: i64=sys_openat_wr(corpus_path,0x1a4)
200 let docmap_fd: i64=sys_openat_wr(docmap_path,0x1a4)
201 let man_fd: i64=sys_openat_append(man_path,0x1a4)
202 let reg_fd: i64=sys_openat_append(reg_path,0x1a4)
203 let flag_fd: i64=sys_openat_append(flag_path,0x1a4)
204
205 // ---- walk knowledge/fetched ----
206 let fd: i64=nx_open_rd("knowledge/fetched" as *u8)
207 if fd < 0 { rg_puts(" FAIL open knowledge/fetched\n" as *u8); sys_exit(1); return 1 }
208 let dirbuf: *u8=sys_mmap(RG_DIR_BUF+64)
209 let pathbuf: *u8=sys_mmap(RG_PATH_CAP)
210 let topcur: *u8=sys_mmap(512)
211 let lenbox: *i64=sys_mmap(16) as *i64
212 let d: *NxDirent=sys_mmap(NX_DIRENT_BYTES) as *NxDirent
213 var added: i64=0
214 var dup_skip: i64=0
215 var xtopic: i64=0
216 var processed: i64=0
217 var done: i64=0
218 while done==0 {
219 let n: i64=nx_dirent_read(fd, dirbuf, RG_DIR_BUF)
220 if n<=0 { done=1 }
221 else {
222 var off: i64=0
223 while off < n {
224 off=nx_dirent_iter(dirbuf, off, n, d)
225 if off<0 { off=n }
226 else {
227 if d.dtype != NX_DT_DIR {
228 let nlen: i64=nx_dirent_name_len(d)
229 if rg_indexable(d.name, nlen)==1 {
230 if processed < RG_BATCH_MAX {
231 var po: i64=rg_cat(pathbuf,0,"knowledge/fetched/" as *u8)
232 po=rg_catbuf(pathbuf,po,d.name,nlen)
233 pathbuf[po]=0 as u8
234 if rg_seen(man, mlen, pathbuf, po)==0 {
235 let buf: *u8=sys_read_file(pathbuf, lenbox)
236 if buf != 0 as *u8 {
237 processed=processed+1
238 let clen: i64=rg_topic(d.name, nlen, topcur) // this doc's topic
239 let fp: i64=rg_fp(buf, lenbox[0])
240 // is this content already registered? (content-dedup)
241 var hit: i64=0-1
242 var k: i64=0
243 while k < nreg { if fps[k]==fp { hit=k } k=k+1 }
244 // record manifest (seen) + registry line always
245 sys_write(man_fd, pathbuf, po); sys_write(man_fd, "\n" as *u8, 1)
246 let rl: *u8=sys_mmap(RG_PATH_CAP+128); var rp: i64=0
247 rp=rg_catnum(rl,rp,fp); rl[rp]=32 as u8; rp=rp+1
248 rp=rg_catbuf(rl,rp,topcur,clen); rl[rp]=32 as u8; rp=rp+1
249 rp=rg_catbuf(rl,rp,pathbuf,po); rl[rp]=10 as u8; rp=rp+1
250 sys_write(reg_fd, rl, rp)
251 if hit >= 0 {
252 // DUPLICATE content -> do NOT index; flag cross-topic if topic differs
253 dup_skip=dup_skip+1
254 let ct_off: i64=toff[hit]; let ct_len: i64=tlen[hit]
255 if rg_streq(((topbuf as i64)+ct_off) as *u8, ct_len, topcur, clen)==0 {
256 xtopic=xtopic+1
257 let fl: *u8=sys_mmap(RG_PATH_CAP+256); var fpos: i64=0
258 fpos=rg_cat(fl,fpos,"CROSS-TOPIC content=" as *u8); fpos=rg_catnum(fl,fpos,fp)
259 fpos=rg_cat(fl,fpos," topics=" as *u8); fpos=rg_catbuf(fl,fpos,((topbuf as i64)+ct_off) as *u8,ct_len)
260 fl[fpos]=43 as u8; fpos=fpos+1; fpos=rg_catbuf(fl,fpos,topcur,clen)
261 fpos=rg_cat(fl,fpos," same_place=" as *u8); fpos=rg_catbuf(fl,fpos,pathbuf,po); fl[fpos]=10 as u8; fpos=fpos+1
262 sys_write(flag_fd, fl, fpos)
263 }
264 }
265 else {
266 // NEW content -> index into shard + register fp (canonical)
267 let wrote: i64=rg_emit(corpus_fd, buf, lenbox[0])
268 if wrote==1 {
269 sys_write(docmap_fd, pathbuf, po); sys_write(docmap_fd, "\n" as *u8, 1)
270 if nreg < RG_REG_CAP { fps[nreg]=fp; toff[nreg]=tpos; tlen[nreg]=clen; tpos=rg_catbuf(topbuf,tpos,topcur,clen); nreg=nreg+1 }
271 added=added+1
272 }
273 }
274 }
275 }
276 }
277 }
278 }
279 }
280 }
281 }
282 }
283 sys_close(fd); sys_close(corpus_fd); sys_close(docmap_fd); sys_close(man_fd); sys_close(reg_fd); sys_close(flag_fd)
284
285 rg_puts(" shard="); rg_num(shard_n); rg_puts(" new="); rg_num(added)
286 rg_puts(" dup_skipped="); rg_num(dup_skip); rg_puts(" cross_topic="); rg_num(xtopic)
287 rg_puts(" unique_total="); rg_num(nreg); rg_puts("\n" as *u8)
288
289 // ---- growth ledger (ALWAYS) ----
290 let lfd: i64=sys_openat_append(led_path, 0x1a4)
291 if lfd >= 0 {
292 let line: *u8=sys_mmap(256); var lp: i64=0
293 lp=rg_cat(line,lp,"shard=" as *u8); lp=rg_catnum(line,lp,shard_n)
294 lp=rg_cat(line,lp," new=" as *u8); lp=rg_catnum(line,lp,added)
295 lp=rg_cat(line,lp," dup=" as *u8); lp=rg_catnum(line,lp,dup_skip)
296 lp=rg_cat(line,lp," xtopic=" as *u8); lp=rg_catnum(line,lp,xtopic)
297 lp=rg_cat(line,lp," unique=" as *u8); lp=rg_catnum(line,lp,nreg)
298 lp=rg_cat(line,lp,"\n" as *u8)
299 sys_write(lfd, line, lp); sys_close(lfd)
300 }
301
302 if added == 0 {
303 rg_puts(" NO-NEW-CONTENT this run (processed="); rg_num(processed); rg_puts(" dup_skipped="); rg_num(dup_skip)
304 rg_puts("; prior data intact, dedup held)\n" as *u8)
305 sys_exit(0); return 0
306 }
307
308 // ---- build + persist the shard from NEW content only (bounded = OOM-safe), self-check ----
309 let idx: *NxInvIndex=nx_inv_build_from_jsonl(corpus_path)
310 if idx == 0 as *NxInvIndex { rg_puts(" FAIL shard build\n" as *u8); sys_exit(1); return 1 }
311 let sv: i64=nx_inv_save(idx, idx_path)
312 rg_puts(" shard indexed rows="); rg_num(idx.n_rows); rg_puts(" persisted="); rg_num(sv); rg_puts("\n" as *u8)
313
314 rg_puts(" GROW-OK: +"); rg_num(added); rg_puts(" new content, "); rg_num(dup_skip)
315 rg_puts(" dup skipped, "); rg_num(xtopic); rg_puts(" cross-topic links flagged (unique_before="); rg_num(unique_before)
316 rg_puts(" unique_after="); rg_num(nreg); rg_puts(", append-only, no data lost)\n" as *u8)
317 sys_exit(0); return 0
318}