nx_research_engine.nx source
↩ module page · 351 lines · 18476 B
1// nx_research_engine.nx -- the SHARED sovereign research fetch->extract->bank ENGINE. Rule-15 DRY: ONE copy of the
2// fetch loop that was copy-pasted into every nx_*_research_fetch organ. Topic organs (nx_arith_research, ...) stay
3// THIN: they COMPOSE this engine and declare their STRUCTURED sources (sections + purpose) -- no copy-pasted loop,
4// no flat TSV. Functionality is a strict SUPERSET of the old fetchers:
5// PRESERVED: idempotent have-skip, sovereign TLS-1.3 fetch, status print, GZIP-guard, raw archive, readable count,
6// descriptive SECTIONS (rf_section).
7// ADDED: HTML -> clean plain text -> knowledge/library/<name>.txt (so nx_library_harvest_v2 banks it).
8// 100% sovereign (own TLS, nx_cc->nxasm, no curl/wget/gcc). license_tier: ORIGINAL
9import "nx_syscalls.nx"
10import "nx_itoa_lib.nx" // shared MSB-first emitter (zero-alloc)
11import "nx_x509_trust_store.nx"
12import "nx_trust_store_load_from_certdata.nx"
13import "nx_https_fetch_follow.nx"
14import "nx_research_ka.nx"
15import "nx_fetch_any.nx" // EAT-THE-DEBT 2026-07-14: intelligent cascade + classify + always-inform
16import "nx_gzip_wrap.nx" // we advertise gzip, so we must be able to decode it
17const RF_GZ_MAX_OUT: i64 = 16777216 // 16 MiB inflated cap per document
18const RF_MAGIC_2026: i64 = 2026
19const RF_MAGIC_4194304: i64 = 4194304
20const RF_MAGIC_2048: i64 = 2048
21const RF_MAGIC_16384: i64 = 16384
22const RF_MAGIC_16383: i64 = 16383
23const RF_MAGIC_1024: i64 = 1024
24const RF_MAGIC_4096: i64 = 4096
25const RF_MAGIC_8192: i64 = 8192
26
27func rf_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
28// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
29// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
30// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
31// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
32func rf_putn(v: i64) -> i64 { nxi_out(v); return 0 }
33func rf_have(path: *u8) -> i64 { let fd: i64 = sys_openat_rd(path); if fd < 0 { return 0 } sys_close(fd); return 1 }
34func rf_scat(dst: *u8, o: i64, s: *u8) -> i64 { var i: i64 = 0; while s[i] != (0 as u8) { dst[o + i] = s[i]; i = i + 1 } return o + i }
35func rf_bpath(dst: *u8, dir: *u8, name: *u8, suf: *u8) -> i64 {
36 var o: i64 = rf_scat(dst, 0, dir); o = rf_scat(dst, o, name); o = rf_scat(dst, o, suf); dst[o] = 0 as u8; return o
37}
38func rf_save(path: *u8, buf: *u8, n: i64) -> i64 {
39 let fd: i64 = sys_openat_wr(path, 0x1a4)
40 if fd < 0 { return 0 }
41 var off: i64 = 0
42 while off < n { let w: i64 = sys_write(fd, ((buf as i64) + off) as *u8, n - off); if w <= 0 { sys_close(fd); return 0 } off = off + w }
43 sys_close(fd); return 1
44}
45
46// case-insensitive compare src[off..off+litlen) == lit (lit lowercase)
47func rf_tag_eq(src: *u8, off: i64, n: i64, lit: *u8, litlen: i64) -> i64 {
48 if off + litlen > n { return 0 }
49 var i: i64 = 0
50 while i < litlen {
51 var c: i64 = src[off + i] as i64
52 if c >= 0x41 { if c <= 0x5a { c = c + 0x20 } }
53 if c != (lit[i] as i64) { return 0 }
54 i = i + 1
55 }
56 return 1
57}
58
59// minimal HTML -> plain text for the corpus: drop <...> tags, suppress <script>/<style> content, collapse ws.
60func rf_strip(src: *u8, n: i64, out: *u8, cap: i64) -> i64 {
61 var op: i64 = 0
62 var i: i64 = 0
63 var last_ws: i64 = 1
64 var suppress: i64 = 0
65 while i < n {
66 let c: i64 = src[i] as i64
67 if c == 0x3c {
68 var nameoff: i64 = i + 1
69 var isclose: i64 = 0
70 if nameoff < n { if (src[nameoff] as i64) == 0x2f { isclose = 1; nameoff = nameoff + 1 } }
71 var iss: i64 = 0
72 if rf_tag_eq(src, nameoff, n, "script" as *u8, 6) == 1 { iss = 1 }
73 if rf_tag_eq(src, nameoff, n, "style" as *u8, 5) == 1 { iss = 1 }
74 if iss == 1 { if isclose == 1 { suppress = 0 } else { suppress = 1 } }
75 i = i + 1
76 var g: i64 = 0
77 while g == 0 { if i >= n { g = 1 } else { if (src[i] as i64) == 0x3e { g = 1 } else { i = i + 1 } } }
78 if i < n { i = i + 1 }
79 if last_ws == 0 { if op < cap { out[op] = 0x20 as u8; op = op + 1; last_ws = 1 } }
80 } else {
81 if suppress == 1 { i = i + 1 }
82 else {
83 var isws: i64 = 0
84 if c == 0x20 { isws = 1 }
85 if c == 0x09 { isws = 1 }
86 if c == 0x0a { isws = 1 }
87 if c == 0x0d { isws = 1 }
88 if isws == 1 {
89 if last_ws == 0 { if op < cap { out[op] = 0x20 as u8; op = op + 1; last_ws = 1 } }
90 } else {
91 if op < cap { out[op] = c as u8; op = op + 1; last_ws = 0 }
92 }
93 i = i + 1
94 }
95 }
96 }
97 return op
98}
99
100// ---- the engine API the thin topic organs compose ----
101
102// load the sovereign trust store + ensure dirs. Returns the store, or 0 on failure.
103func rf_init() -> *TrustStore {
104 let r: i64 = nx_trust_store_load_from_certdata("data/mozilla_certdata.txt" as *u8, 512, RF_MAGIC_4194304)
105 if r <= 0 { return 0 as *TrustStore }
106 sys_mkdir("knowledge/fetched" as *u8, 0x1ed)
107 sys_mkdir("knowledge/library" as *u8, 0x1ed)
108 return r as *TrustStore
109}
110
111// print a research SECTION header (the structure/intelligence the old fetchers carried -- preserved).
112func rf_section(s: *u8) -> i64 { rf_puts(" == " as *u8); rf_puts(s); rf_puts(" ==\n" as *u8); return 0 }
113
114// GZIP magic detection (0x1f 0x8b) -- the guard the old fetchers carried; a gzip body must be skipped, never
115// banked as garbage. A tested function (nx_research_engine_gate) rather than an inline check.
116func rf_is_gzip(buf: *u8, n: i64) -> i64 { if n >= 2 { if (buf[0] as i64) == 0x1f { if (buf[1] as i64) == 0x8b { return 1 } } } return 0 }
117
118// ---- R2/R3: POLITE + BOUNDED-PARALLEL fetch coordination (folded into the engine so ALL researchers inherit it).
119// N lease SLOTS sized to free RAM; each network fetch acquires ANY free slot (O_EXCL; steals a crashed holder's
120// stale slot >30s). N researcher PROCESSES coordinate through the shared slot dir -> at most N cert-verifies at
121// once (no memory thrash/hang); all slots busy -> WAIT politely, never stampede. Tuning consts -> svc-config.
122const RF_OEXCL: i64 = 193
123const RF_MODE: i64 = 420
124const RF_MAX_SLOTS: i64 = 4
125const RF_PER_FETCH_MB: i64 = 256
126const RF_STALE_SEC: i64 = 30
127const RF_WAIT_MS: i64 = 200
128const RF_WAIT_MAX: i64 = 300
129const RF_HOST_DELAY: i64 = 1 // seconds between fetches to the SAME host (polite per-host crawl-delay) -> svc-config
130func rf_sleep_ms(ms: i64) -> i64 { sys_poll(0 as *u8, 0, ms); return 0 }
131func rf_free_mb() -> i64 {
132 let fd: i64 = sys_openat_rd("/proc/meminfo" as *u8)
133 if fd < 0 { return RF_MAGIC_2048 }
134 let b: *u8 = sys_mmap(RF_MAGIC_16384); let n: i64 = sys_read(fd, b, RF_MAGIC_16383); sys_close(fd)
135 let pat: *u8 = "MemAvailable:" as *u8
136 var i: i64 = 0; var found: i64 = 0 - 1
137 while i < n - 13 {
138 var j: i64 = 0; var m: i64 = 1
139 while j < 13 { if b[i+j] != pat[j] { m = 0; j = 13 } else { j = j + 1 } }
140 if m == 1 { found = i + 13; i = n } else { i = i + 1 }
141 }
142 if found < 0 { return RF_MAGIC_2048 }
143 var kb: i64 = 0; var started: i64 = 0; var done: i64 = 0; var p: i64 = found
144 while done == 0 {
145 if p >= n { done = 1 } else {
146 let c: i64 = b[p] as i64
147 var isdig: i64 = 0
148 if c >= 48 { if c <= 57 { isdig = 1 } }
149 if isdig == 1 { kb = kb * 10 + (c - 48); started = 1; p = p + 1 }
150 else { if started == 1 { done = 1 } else { p = p + 1 } }
151 }
152 }
153 return kb / RF_MAGIC_1024
154}
155func rf_nslots() -> i64 {
156 let mb: i64 = rf_free_mb()
157 var n: i64 = mb / RF_PER_FETCH_MB
158 if n < 1 { n = 1 }
159 if n > RF_MAX_SLOTS { n = RF_MAX_SLOTS }
160 return n
161}
162func rf_slotpath(dir: *u8, idx: i64, out: *u8) -> *u8 {
163 var o: i64 = 0; var i: i64 = 0
164 while dir[i] != (0 as u8) { out[o] = dir[i]; o = o + 1; i = i + 1 }
165 out[o] = 47 as u8; o = o + 1
166 out[o] = 115 as u8; o = o + 1
167 out[o] = (48 + idx) as u8; o = o + 1
168 out[o] = 46 as u8; o = o + 1
169 out[o] = 108 as u8; o = o + 1
170 out[o] = 107 as u8; o = o + 1
171 out[o] = 0 as u8
172 return out
173}
174// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
175// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
176// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
177// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
178func rf_wint(fd: i64, v: i64) -> i64 { nxi_fd(fd, v); return 0 }
179func rf_rint(path: *u8) -> i64 { let fd: i64=sys_openat_rd(path); if fd<0{return 0} let b: *u8=sys_mmap(64); let n: i64=sys_read(fd,b,63); sys_close(fd); var v: i64=0; var i: i64=0; while i<n { if b[i]>=(48 as u8){if b[i]<=(57 as u8){v=v*10+((b[i] as i64)-48)}} i=i+1 } return v }
180func rf_unlink(path: *u8) -> i64 { return __syscall(263, AT_FDCWD, path as i64, 0, 0, 0, 0) }
181func rf_slot_lock(path: *u8) -> i64 {
182 let fd: i64 = __syscall(SYS_OPENAT, AT_FDCWD, path as i64, RF_OEXCL, RF_MODE, 0, 0)
183 if fd >= 0 { rf_wint(fd, sys_now_realtime_sec()); return fd }
184 let held: i64 = rf_rint(path)
185 if held > 0 { if sys_now_realtime_sec() - held > RF_STALE_SEC {
186 rf_unlink(path)
187 let fd2: i64 = __syscall(SYS_OPENAT, AT_FDCWD, path as i64, RF_OEXCL, RF_MODE, 0, 0)
188 if fd2 >= 0 { rf_wint(fd2, sys_now_realtime_sec()); return fd2 }
189 } }
190 return 0 - 1
191}
192func rf_slot_acquire(dir: *u8, n: i64, idx_out: *i64) -> i64 {
193 let pbuf: *u8 = sys_mmap(512)
194 var wait: i64 = 0
195 while wait < RF_WAIT_MAX {
196 var i: i64 = 0
197 while i < n {
198 rf_slotpath(dir, i, pbuf)
199 let fd: i64 = rf_slot_lock(pbuf)
200 if fd >= 0 { idx_out[0] = i; return fd }
201 i = i + 1
202 }
203 rf_sleep_ms(RF_WAIT_MS); wait = wait + 1
204 }
205 return 0 - 1
206}
207func rf_slot_release(dir: *u8, idx: i64, fd: i64) -> i64 {
208 let pbuf: *u8 = sys_mmap(512); rf_slotpath(dir, idx, pbuf)
209 sys_close(fd); rf_unlink(pbuf); return 0
210}
211// ---- R2b/R6b per-host politeness + resilience: host-keyed state files under research_hosts/<host>.<suf> ----
212const RF_CB_THRESHOLD: i64 = 3 // consecutive fails to TRIP the circuit breaker
213const RF_CB_COOLDOWN: i64 = 60 // seconds a tripped host is fast-failed (don't hammer a down site)
214func rf_wfile(path: *u8, v: i64) -> i64 {
215 let fd: i64 = __syscall(SYS_OPENAT, AT_FDCWD, path as i64, 577, RF_MODE, 0, 0) // O_WRONLY|O_CREAT|O_TRUNC
216 if fd >= 0 { rf_wint(fd, v); sys_close(fd) }
217 return 0
218}
219// build "knowledge/status/research_hosts/<sanitized-host>.<suf>" into out (host parsed after "://", until '/')
220func rf_host_key(url: *u8, suf: *u8, out: *u8) -> i64 {
221 var i: i64 = 0; var schemeend: i64 = 0; var d1: i64 = 0
222 while d1 == 0 {
223 if url[i] == (0 as u8) { d1 = 1 }
224 else { if url[i]==(58 as u8) { if url[i+1]==(47 as u8) { if url[i+2]==(47 as u8) { schemeend = i+3; d1 = 1 } } } if d1==0 { i=i+1 } }
225 }
226 let pre: *u8 = "knowledge/status/research_hosts/" as *u8
227 var o: i64 = 0; var j: i64 = 0
228 while pre[j] != (0 as u8) { out[o]=pre[j]; o=o+1; j=j+1 }
229 var h: i64 = schemeend; var d2: i64 = 0
230 while d2 == 0 {
231 let c: i64 = url[h] as i64
232 if c == 0 { d2 = 1 } else { if c == 47 { d2 = 1 } else {
233 var ok: i64 = 0
234 if c>=48 { if c<=57 { ok=1 } }
235 if c>=97 { if c<=122 { ok=1 } }
236 if c>=65 { if c<=90 { ok=1 } }
237 if ok==1 { out[o]=url[h] } else { out[o]=95 as u8 }
238 o=o+1; h=h+1
239 } }
240 }
241 out[o]=46 as u8; o=o+1
242 var s: i64 = 0
243 while suf[s] != (0 as u8) { out[o]=suf[s]; o=o+1; s=s+1 }
244 out[o]=0 as u8
245 return o
246}
247// R2b PER-HOST CRAWL-DELAY (polite): ensure >= RF_HOST_DELAY sec between fetches to the SAME host (cross-process).
248func rf_host_gate(url: *u8) -> i64 {
249 let fn: *u8 = sys_mmap(512); rf_host_key(url, "t" as *u8, fn)
250 sys_mkdir("knowledge/status/research_hosts" as *u8, 0x1ed)
251 let last: i64 = rf_rint(fn)
252 let now: i64 = sys_now_realtime_sec()
253 if last > 0 { let gap: i64 = now - last; if gap < RF_HOST_DELAY { rf_sleep_ms((RF_HOST_DELAY - gap) * 1000) } }
254 rf_wfile(fn, sys_now_realtime_sec())
255 return 0
256}
257// R6b CIRCUIT BREAKER: is host tripped (fast-fail window active)? <host>.cbt = trip-until epoch (0 = closed).
258func rf_circuit_open(url: *u8) -> i64 {
259 let fn: *u8 = sys_mmap(512); rf_host_key(url, "cbt" as *u8, fn)
260 let trip: i64 = rf_rint(fn)
261 if trip > 0 { if sys_now_realtime_sec() < trip { return 1 } }
262 return 0
263}
264// record the outcome: success -> reset; failure -> count, and TRIP for RF_CB_COOLDOWN after RF_CB_THRESHOLD in a row.
265func rf_circuit_record(url: *u8, ok: i64) -> i64 {
266 let fnf: *u8 = sys_mmap(512); rf_host_key(url, "cbf" as *u8, fnf)
267 let fnt: *u8 = sys_mmap(512); rf_host_key(url, "cbt" as *u8, fnt)
268 if ok == 1 { rf_wfile(fnf, 0); rf_wfile(fnt, 0); return 0 }
269 var f: i64 = rf_rint(fnf); f = f + 1
270 if f >= RF_CB_THRESHOLD { rf_wfile(fnt, sys_now_realtime_sec() + RF_CB_COOLDOWN); rf_wfile(fnf, 0) }
271 else { rf_wfile(fnf, f) }
272 return 0
273}
274
275// fetch ONE source -> raw archive + clean library .txt. idempotent (skip if already banked; reuse raw archive),
276// GZIP-guarded (skip gzip bodies -- PRESERVED from the old fetchers). Returns 1 iff a clean .txt is banked.
277func rf_fetch_bank(url: *u8, name: *u8, store: *TrustStore, out: *u8, cap: i64) -> i64 {
278 let rawpath: *u8 = sys_mmap(RF_MAGIC_4096); rf_bpath(rawpath, "knowledge/fetched/" as *u8, name, ".raw" as *u8)
279 let txtpath: *u8 = sys_mmap(RF_MAGIC_4096); rf_bpath(txtpath, "knowledge/library/" as *u8, name, ".txt" as *u8)
280 let lenbox: *i64 = sys_mmap(16) as *i64
281 if rf_have(txtpath) == 1 { rf_puts(" "); rf_puts(name); rf_puts(" [banked-skip]\n" as *u8); return 1 }
282 var raw: *u8 = 0 as *u8
283 var n: i64 = 0
284 if rf_have(rawpath) == 1 {
285 raw = sys_read_file(rawpath, lenbox); n = lenbox[0]
286 rf_puts(" "); rf_puts(name); rf_puts(" [raw cached]")
287 } else {
288 // R6b CIRCUIT BREAKER: skip a recently-failing host (fast-fail, no slot -- don't hammer a down site).
289 if rf_circuit_open(url) == 1 { rf_puts(" "); rf_puts(name); rf_puts(" [circuit-open skip]\n" as *u8); return 0 }
290 // R2+R3: acquire a resource-sized concurrency SLOT (wait politely if all busy).
291 let sdir: *u8 = "knowledge/status/research_slots" as *u8
292 sys_mkdir(sdir, 0x1ed)
293 rf_host_gate(url) // R2b: polite per-host crawl-delay (no slot held while waiting)
294 let nsl: i64 = rf_nslots()
295 let sidx: *i64 = sys_mmap(16) as *i64
296 let sfd: i64 = rf_slot_acquire(sdir, nsl, sidx)
297 let status: *i64 = sys_mmap(8) as *i64
298 // EAT-THE-DEBT (operator 2026-07-14): route through nx_fetch_any -- the INTELLIGENT dispatcher that
299 // CASCADES fetchers (sovereign TLS-1.3/1.2 -> browser spoof) and CLASSIFIES the result, so a Cloudflare
300 // challenge or an empty is NEVER banked as if it were real research (the old rf_fetch_ka banked 200-OK
301 // challenge HTML). nx_fetch_any cascades strategies internally; one polite retry on a hard failure.
302 let attbuf: *u8 = sys_mmap(RF_MAGIC_8192)
303 n = nx_fetch_any(url, store, out, cap, 6, status, 0 as *u8, 0, attbuf, RF_MAGIC_8192)
304 if n <= 0 { rf_sleep_ms(500); n = nx_fetch_any(url, store, out, cap, 6, status, 0 as *u8, 0, attbuf, RF_MAGIC_8192) }
305 if sfd >= 0 { rf_slot_release(sdir, sidx[0], sfd) }
306 var cbok: i64 = 0
307 if n > 0 { cbok = 1 }
308 rf_circuit_record(url, cbok) // R6b: feed the breaker (reset on success, count on fail)
309 rf_sleep_ms(150) // polite inter-fetch delay
310 rf_puts(" "); rf_puts(name); rf_puts(" status="); rf_putn(status[0])
311 if n <= 0 {
312 // ALWAYS-INFORM (operator: "dont return nothing without informing the user"): bank a <name>.fail
313 // sidecar carrying the per-strategy attempts log, so absence becomes DATA the census/adversary/critic
314 // can SEE -- never a silent empty file.
315 let failpath: *u8 = sys_mmap(RF_MAGIC_4096); rf_bpath(failpath, "knowledge/library/" as *u8, name, ".fail" as *u8)
316 var fl: i64 = 0
317 while attbuf[fl] != (0 as u8) { fl = fl + 1 }
318 rf_save(failpath, attbuf, fl)
319 rf_puts(" FETCH-FAILED -> banked " as *u8); rf_puts(name); rf_puts(".fail\n" as *u8)
320 rf_puts(" " as *u8); rf_puts(attbuf)
321 return 0
322 }
323 // INFLATE -- do not discard. nx_codec_caps DERIVES Accept-Encoding from the decoders that exist
324 // and now advertises gzip, so servers legitimately compress. Throwing the body away made the
325 // request a lie in the OTHER direction: we asked for gzip, got 200, and banked ZERO documents.
326 // Measured 2026-08-01: arch 0/14, archive 0/18, adversarial 0/20 -- which is why
327 // knowledge/library did not exist at all and the librarian domain read RED for "no corpus".
328 // The decoder (nx_gzip_wrap) has been in-tree the whole time. THE HEADER MUST EQUAL THE
329 // CAPABILITY, IN BOTH DIRECTIONS.
330 if rf_is_gzip(out, n) == 1 {
331 let gr: *NxGzipResult = nx_gzip_inflate(out, n, RF_GZ_MAX_OUT)
332 let gerr: i64 = gr.error_code
333 if gerr != 0 { rf_puts(" [GZIP-FAIL code=" as *u8); rf_putn(gerr); rf_puts("]\n" as *u8); return 0 }
334 let gsz: i64 = gr.output_size
335 if gsz <= 0 { rf_puts(" [GZIP-EMPTY]\n" as *u8); return 0 }
336 raw = gr.output_data
337 n = gsz
338 rf_save(rawpath, raw, n)
339 rf_puts(" [gunzip]" as *u8)
340 } else {
341 rf_save(rawpath, out, n); raw = out
342 }
343 }
344 let txt: *u8 = sys_mmap(cap)
345 let tn: i64 = rf_strip(raw, n, txt, cap)
346 if rf_save(txtpath, txt, tn) == 0 { rf_puts(" TXT-FAIL\n" as *u8); return 0 }
347 rf_puts(" raw="); rf_putn(n); rf_puts(" -> lib.txt="); rf_putn(tn); rf_puts(" BANKED\n" as *u8)
348 return 1
349}
350
351func main() -> i64 { return 0 }