nx_lean_dir_ingest.nx source
↩ module page · 87 lines · 3444 B
1// nx_lean_dir_ingest.nx -- ingest every Lean decl in the corpus at
2// nxc2/_offc/lean_corpus.txt (assembled by nx_ingest_dir.sh from a
3// caller-pointed directory) and emit shards under /tmp/nx_ingest_lean-*.
4//
5// Thin driver: read corpus -> nx_lean_ingest_corpus -> offer through
6// nx_ingest_runner via nx_lean_stream_offer_all -> close. The
7// per-source bridge does the JSONL row layout; the runner handles
8// dedup / disk_budget / shard rotation / checkpoint / watchdog.
9//
10// "Point at X" capability: only the bash side (nx_ingest_dir.sh) needs
11// to know which directory to read from. Everything else is identical
12// across the 5 source types.
13//
14// genealogy_id: substrate_self_hosting_synth_2026_05_14 + point_at_x_ingester
15// lineage_id: per_source_dir_driver
16
17// nx_safety_envelope:
18// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
19// sil_target: SIL1
20// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
21// verdict: NOT_YET_EVALUATED
22
23import "nx_syscalls.nx"
24import "nx_runtime.nx"
25import "nx_tier.nx"
26import "nx_lean_ingest.nx"
27import "nx_ingest_runner.nx"
28import "nx_lean_stream_ingest.nx"
29import "nx_bloom_capacity.nx"
30
31const NX_LEAN_DIR_DISK_BUDGET: nx_size = 1073741824 // 1 GiB
32const NX_LEAN_DIR_SHARD_BYTES: nx_size = 104857600 // 100 MiB
33const NX_LEAN_DIR_BLOOM_CAPACITY: nx_int = 1000000 // size for 1M items
34const NX_LEAN_DIR_WATCHDOG_MS: nx_int = 60000
35
36func main() -> nx_exit {
37 let corpus_path: *u8 = "nxc2/_offc/lean_corpus.txt" as *u8
38 let out_prefix: *u8 = "/tmp/nx_ingest_lean" as *u8
39 let ckpt_path: *u8 = "/tmp/nx_ingest_lean.checkpoint" as *u8
40
41 let corpus_len_p: *nx_size = (sys_mmap(NX_SIZEOF_NX_SIZE)) as *nx_size
42 corpus_len_p[0] = 0
43 let corpus: *u8 = sys_read_file(corpus_path, corpus_len_p)
44 if (corpus as nx_size) == 0 {
45 sys_write(NX_FD_STDOUT, "nx_lean_dir_ingest: cannot read corpus\n" as *u8, 39)
46 return 1
47 }
48 let corpus_n: nx_size = corpus_len_p[0]
49
50 sys_write(NX_FD_STDOUT, "nx_lean_dir_ingest: corpus bytes=" as *u8, 33)
51 print_i64(corpus_n as i64)
52 sys_write(NX_FD_STDOUT, "\n" as *u8, 1)
53
54 let db: *LeanDb = nx_lean_ingest_corpus(corpus, corpus_n as i64)
55 sys_write(NX_FD_STDOUT, "nx_lean_dir_ingest: decls parsed=" as *u8, 33)
56 print_i64(db.n_decls as i64)
57 sys_write(NX_FD_STDOUT, "\n" as *u8, 1)
58
59 let run: *NxIngestRun = nx_ingest_run_new(
60 out_prefix,
61 NX_LEAN_DIR_DISK_BUDGET,
62 NX_LEAN_DIR_SHARD_BYTES,
63 NX_LEAN_DIR_BLOOM_CAPACITY,
64 NX_BLOOM_PROFILE_1PCT,
65 ckpt_path,
66 NX_LEAN_DIR_WATCHDOG_MS)
67 if run == (0 as *NxIngestRun) {
68 sys_write(NX_FD_STDOUT, "nx_lean_dir_ingest: runner alloc failed\n" as *u8, 40)
69 return 2
70 }
71
72 let row_buf: *u8 = sys_mmap(NX_BUF_HUGE)
73 let n_emitted: nx_int = nx_lean_stream_offer_all(db, run, row_buf)
74
75 sys_write(NX_FD_STDOUT, "nx_lean_dir_ingest: emitted=" as *u8, 28)
76 print_i64(n_emitted as i64)
77 sys_write(NX_FD_STDOUT, " duplicate=" as *u8, 11)
78 print_i64(nx_ingest_run_n_duplicate(run) as i64)
79 sys_write(NX_FD_STDOUT, " budget_halt=" as *u8, 13)
80 print_i64(nx_ingest_run_n_budget_halt(run) as i64)
81 sys_write(NX_FD_STDOUT, " shards=" as *u8, 8)
82 print_i64(nx_ingest_run_n_shards(run) as i64)
83 sys_write(NX_FD_STDOUT, "\n" as *u8, 1)
84
85 nx_ingest_run_close(run)
86 return 0
87}