nx_bright_rewrite_lib.nx source
↩ module page · 612 lines · 34055 B
1// nx_bright_rewrite_lib.nx -- R0l: the SOVEREIGN reasoning-rewrite run for BRIGHT as a LIBRARY (search R0l, 2026-09-16).
2//
3// WHY. The referee replay (R0o) proved the rewrite lever is real at 1.5B: the authors' own TongSearch-QR-1.5B
4// rewrites score 229 permil through the estate's plain BM25 against the 149 floor. That number was produced by a
5// query another model wrote, so it can never place the estate on the board. This library produces the rewrites on
6// the estate's OWN engine (nx_nofloat_serve_core, the Q8_0 GGUF twin of the MIT weights) with the paper's own
7// prompt, so the queries it writes are the estate's system and nx_beir_eval may score them under the pre-declared
8// accept rule: within 30 permil of the 229 replay, or the port is not the arm.
9//
10// TWO ENGINE DOORS, ONE RUN (operator 2026-09-16: "through mcp and apis and primitives and workflows and agents not
11// wsl ... from the first byte up"):
12// in-process gguf=<path> the model is loaded into this process (nsv_init_i8) -- the laptop lane.
13// remote engine=a.b.c.d:port every prompt is POSTed to a sovereign nx_nofloat_serve on the LAN (POST /gen,
14// the same JSON contract the app page and nx_engine_dispatch use), so a NAS plan
15// row drives a compute node's engine and the NAS never runs the model. The
16// remote door composes nx_http_client (request), nx_http_response_parse
17// (reply) and the serve core's own JSON helpers (nsv_jesc / nsv_jkey /
18// nsv_jstr / nsv_jint) -- one escaper, one key finder, no second ruler.
19// A per-prompt REFUSAL (the engine answered 400 with an err code: too long for its window) is written as the
20// ORIGINAL query and counted as passthrough, exactly as in-process. An ENGINE FAILURE (unreachable, no reply,
21// an unparseable reply, or err=1 which means the request itself was malformed) is NOT a refusal: it ABORTS the
22// split (state ENGINE-DOWN) so a dead node can never manufacture a "complete" file of original queries.
23//
24// IDEMPOTENT RE-RUNS, so a clock agent may call this every beat:
25// * a split whose final queries.<out>.tsv already exists is skipped as COMPLETE-EXISTING (no engine call);
26// * a split is run under a lease (nx_lease_lib, name brightrewrite-<out>-<split>, ttl lease_ttl); a second
27// runner on the same tree reports LEASED-ELSEWHERE and moves on, so two runners take disjoint splits;
28// * rows already in the .part file are skipped by count (the same resume as before);
29// * a run whose every written row was a passthrough does NOT rename to complete (ALL-PASSTHROUGH): an engine
30// that refuses every prompt has produced nothing, and a file of originals must never be scored as rewrites.
31// Per split it appends <root>/<split>/queries.<out>.tsv.part one row per query as each rewrite lands and renames
32// the part to queries.<out>.tsv only when every query of the split has a row. limit>0 is a SMOKE run: it stops
33// after that many queries per split and never renames, so a smoke file can never be mistaken for a full one.
34// Progress: <root>/../../status/bright_rewrite.<out>.progress (or progress=<path>) is truncate-written after every
35// query (a ts that stops moving while the process lives is a hung run, not a slow one).
36// license_tier: ORIGINAL. No hw writes (Rule 26). LIB (no main).
37import "nx_syscalls.nx"
38import "nx_nofloat_serve_core.nx"
39import "nx_http_client.nx"
40import "nx_http_response_parse.nx"
41import "nx_lease_lib.nx"
42
43const BR_STDOUT: i64 = 1
44const BR_NSPLITS: i64 = 12
45const BR_PATH_CAP: i64 = 1024
46const BR_NUM_CAP: i64 = 32
47const BR_OUT_CAP: i64 = 65536
48const BR_OUT_USE: i64 = 65000
49const BR_GP_SLOTS: i64 = 16
50const BR_META_SLOTS: i64 = 8
51const BR_ST_SLOTS: i64 = 12
52const BR_CFG_SLOTS: i64 = 12
53const BR_ENG_SLOTS: i64 = 8
54const BR_I64: i64 = 8
55const BR_MAXNEW_DEFAULT: i64 = 512 // NSV_MAXNEW in the serve core: a row may ask for less, never more
56const BR_MODE_I8: i64 = 1
57const BR_TEMP_GREEDY: i64 = 0
58const BR_TOPP_PERMIL: i64 = 1000
59const BR_TOPK_DEFAULT: i64 = 64 // the serve daemon's own default; top_k 0 keeps only the argmax, so a sampled run silently decodes GREEDY (measured 2026-09-16: the sampled row was byte-identical to the greedy one)
60const BR_SEED: i64 = 1
61const BR_CHATML: i64 = 1
62const BR_NOSTREAM: i64 = 0
63const BR_NOFD: i64 = 0 - 1
64const BR_TAB: i64 = 9
65const BR_NL: i64 = 10
66const BR_CR: i64 = 13
67const BR_SPACE: i64 = 32
68const BR_SYS_OPENAT: i64 = 257
69const BR_AT_FDCWD: i64 = 0 - 100
70const BR_O_APPEND_CREATE: i64 = 1089 // O_WRONLY|O_CREAT|O_APPEND
71const BR_FILE_MODE: i64 = 420
72// the remote door's wait for ONE reply is derived, not picked: max_new tokens at the slowest host this engine was
73// measured on (the NAS at 772 ms/token, 2026-09-16) doubled, plus the model's prompt walk -- a loaded box pays
74// what it pays, a dead one is named inside the budget
75const BR_ENGINE_TOKEN_MS_BOUND: i64 = 2000
76const BR_ENGINE_SETUP_S: i64 = 60
77const BR_ENGINE_RESP_CAP: i64 = 135168 // 4096 header reserve + twice BR_OUT_CAP: an escaped reply may double
78const BR_ENGINE_BODY_RESERVE: i64 = 512 // the JSON skeleton around an escaped prompt
79const BR_ENGINE_HOST_CAP: i64 = 24 // "255.255.255.255:65535" NUL
80const BR_HTTP_OK: i64 = 200
81const BR_HTTP_BAD: i64 = 400
82const BR_ENGINE_ERR_MALFORMED: i64 = 1 // the serve's err for a body it could not parse: OUR defect, never a refusal
83// br_engine_gen results below zero
84const BR_ENG_REFUSED: i64 = 0 - 1 // per-prompt refusal (meta[5] carries the engine's err) -> passthrough row
85const BR_ENG_ABORT: i64 = 0 - 2 // unparseable / unexpected reply -> the split stops
86const BR_ENG_UNREACHABLE: i64 = 0 - 3 // connect, write or an empty reply -> the split stops
87// split states (st[8])
88const BR_STATE_PARTIAL: i64 = 0
89const BR_STATE_COMPLETE: i64 = 1
90const BR_STATE_SMOKE: i64 = 2
91const BR_STATE_EXISTING: i64 = 3
92const BR_STATE_LEASED: i64 = 4
93const BR_STATE_ENGINE_DOWN: i64 = 5
94const BR_STATE_ALL_PASSTHROUGH: i64 = 6
95const BR_STATE_LEASE_IO: i64 = 7
96const BR_LEASE_TTL_DEFAULT: i64 = 21600 // 6 h: longer than a split at the slowest measured pace, so a live runner is never stolen from; an orphan after a crash frees itself within a beat
97const BR_LEASE_NAME_PFX: *u8 = "brightrewrite-" as *u8
98const BR_PROMPT_HEAD: *u8 = "Instructions:\n1. Identify the essential problem.\n2. Think step by step to reason and describe what information could be relevant and helpful to address the questions in detail.\n3. Draft an answer with as many thoughts as you have.\nQuery: " as *u8
99const BR_PROMPT_TAIL: *u8 = "\n\n" as *u8
100const BR_PROGRESS_REL: *u8 = "/../../status/bright_rewrite." as *u8
101const BR_PROGRESS_EXT: *u8 = ".progress" as *u8
102const BR_GEN_PATH: *u8 = "/gen" as *u8
103const BR_GEN_CTYPE: *u8 = "application/json" as *u8
104// ---- the ORACLE dialect (2026-09-16) ---------------------------------------------------------------------------
105// engine_kind=llamacpp speaks llama.cpp's POST /completion instead of the sovereign /gen. It exists for ONE reason:
106// the same open weights on the laptop GPU produce the twelve-split rewrites in about an hour where the sovereign CPU
107// engine needs about forty, so the lever's worth is measured TODAY as a BAR (an outside oracle, never the build or
108// deliver path) while the sovereign run lands the number of record. Nothing an oracle writes may be scored as the
109// estate's system: the program REFUSES engine_kind=llamacpp unless out= carries the word oracle, and every receipt
110// names the kind. The prompt is wrapped in the sovereign engine's own ChatML shape (nsv_chatml_ids: user / assistant,
111// no system prompt) so the two doors ask the model the identical question.
112const BR_ENGINE_KIND_NISHI: i64 = 0
113const BR_ENGINE_KIND_LLAMACPP: i64 = 1
114const BR_GEN_PATH_LLAMACPP: *u8 = "/completion" as *u8
115const BR_CHATML_HEAD: *u8 = "<|im_start|>user\n" as *u8
116const BR_CHATML_TAIL: *u8 = "<|im_end|>\n<|im_start|>assistant\n" as *u8
117const BR_PERMIL: i64 = 1000
118const BR_ORACLE_MARK: *u8 = "oracle" as *u8
119
120func br_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
121func br_puts(s: *u8) -> i64 { let n: i64 = br_slen(s); if n > 0 { sys_write(BR_STDOUT, s, n) } return 0 }
122func br_putn(v: i64) -> i64 {
123 let t: *u8 = sys_mmap(BR_NUM_CAP)
124 var m: i64 = v
125 var k: i64 = 0
126 if m < 0 { sys_write(BR_STDOUT, "-" as *u8, 1); m = 0 - m }
127 if m == 0 { t[0] = 48 as u8; k = 1 }
128 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
129 let b: *u8 = sys_mmap(BR_NUM_CAP)
130 var i: i64 = 0
131 while i < k { b[i] = t[k - 1 - i]; i = i + 1 }
132 sys_write(BR_STDOUT, b, k)
133 return 0
134}
135func br_cat(dst: *u8, off: i64, s: *u8) -> i64 { var i: i64 = 0; var o: i64 = off; while s[i] != (0 as u8) { dst[o] = s[i]; o = o + 1; i = i + 1 } dst[o] = 0 as u8; return o }
136func br_catb(dst: *u8, off: i64, s: *u8, n: i64) -> i64 { var i: i64 = 0; while i < n { dst[off + i] = s[i]; i = i + 1 } dst[off + n] = 0 as u8; return off + n }
137func br_catn(dst: *u8, off: i64, v: i64) -> i64 {
138 let t: *u8 = sys_mmap(BR_NUM_CAP)
139 var m: i64 = v
140 var o: i64 = off
141 var k: i64 = 0
142 if m < 0 { dst[o] = 45 as u8; o = o + 1; m = 0 - m }
143 if m == 0 { t[0] = 48 as u8; k = 1 }
144 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
145 while k > 0 { k = k - 1; dst[o] = t[k]; o = o + 1 }
146 dst[o] = 0 as u8
147 return o
148}
149func br_starts(s: *u8, pfx: *u8) -> i64 { var i: i64 = 0; while pfx[i] != (0 as u8) { if s[i] != pfx[i] { return 0 } i = i + 1 } return 1 }
150func br_streq(a: *u8, b: *u8) -> i64 { let n: i64 = br_slen(a); if n != br_slen(b) { return 0 } var i: i64 = 0; while i < n { if a[i] != b[i] { return 0 } i = i + 1 } return 1 }
151func br_atoi(s: *u8) -> i64 { var v: i64 = 0; var i: i64 = 0; while s[i] != (0 as u8) { let c: i64 = s[i] as i64; if c < 48 { return v } if c > 57 { return v } v = v * 10 + (c - 48); i = i + 1 } return v }
152// substring find in buf[0,n): index or -1
153func br_find(buf: *u8, n: i64, needle: *u8) -> i64 {
154 let k: i64 = br_slen(needle)
155 if k == 0 { return 0 - 1 }
156 var i: i64 = 0
157 while i + k <= n {
158 var j: i64 = 0
159 var ok: i64 = 1
160 while j < k { if buf[i + j] != needle[j] { ok = 0; j = k } else { j = j + 1 } }
161 if ok == 1 { return i }
162 i = i + 1
163 }
164 return 0 - 1
165}
166// the twelve BRIGHT splits in nx_beir_eval's order
167func br_split(i: i64) -> *u8 {
168 if i == 0 { return "biology" as *u8 }
169 if i == 1 { return "earth_science" as *u8 }
170 if i == 2 { return "economics" as *u8 }
171 if i == 3 { return "psychology" as *u8 }
172 if i == 4 { return "robotics" as *u8 }
173 if i == 5 { return "stackoverflow" as *u8 }
174 if i == 6 { return "sustainable_living" as *u8 }
175 if i == 7 { return "leetcode" as *u8 }
176 if i == 8 { return "pony" as *u8 }
177 if i == 9 { return "aops" as *u8 }
178 if i == 10 { return "theoremqa_questions" as *u8 }
179 return "theoremqa_theorems" as *u8
180}
181func br_path(dst: *u8, root: *u8, split: *u8, tail: *u8) -> i64 {
182 var o: i64 = br_cat(dst, 0, root)
183 dst[o] = 47 as u8; o = o + 1
184 o = br_cat(dst, o, split)
185 dst[o] = 47 as u8; o = o + 1
186 o = br_cat(dst, o, tail)
187 return o
188}
189// <root>/../../status/bright_rewrite.<out>.progress -- CWD-independent: it hangs off the bright root the caller named
190func br_progress_path(dst: *u8, root: *u8, out: *u8) -> i64 {
191 var o: i64 = br_cat(dst, 0, root)
192 o = br_cat(dst, o, BR_PROGRESS_REL)
193 o = br_cat(dst, o, out)
194 o = br_cat(dst, o, BR_PROGRESS_EXT)
195 return o
196}
197func br_count_lines(buf: *u8, n: i64) -> i64 { var c: i64 = 0; var i: i64 = 0; while i < n { if buf[i] == (BR_NL as u8) { c = c + 1 } i = i + 1 } return c }
198func br_write_all(fd: i64, buf: *u8, n: i64) -> i64 {
199 var off: i64 = 0
200 while off < n { let w: i64 = sys_write(fd, buf + off, n - off); if w <= 0 { return 0 - 1 } off = off + w }
201 return 0
202}
203// one line, NUL-terminated, truncate-written: a ts that stops moving while the process lives is a hung run
204func br_progress(path: *u8, split: *u8, done: i64, of: i64, ngen: i64, ms: i64, pass: i64) -> i64 {
205 let b: *u8 = sys_mmap(BR_PATH_CAP)
206 var o: i64 = br_cat(b, 0, "ts=" as *u8); o = br_catn(b, o, sys_now_ms())
207 o = br_cat(b, o, " split=" as *u8); o = br_cat(b, o, split)
208 o = br_cat(b, o, " done=" as *u8); o = br_catn(b, o, done)
209 o = br_cat(b, o, " of=" as *u8); o = br_catn(b, o, of)
210 o = br_cat(b, o, " n_gen=" as *u8); o = br_catn(b, o, ngen)
211 o = br_cat(b, o, " ms=" as *u8); o = br_catn(b, o, ms)
212 o = br_cat(b, o, " passthrough=" as *u8); o = br_catn(b, o, pass)
213 o = br_cat(b, o, " -- truncate-written by nx_bright_rewrite_run after every query\n" as *u8)
214 let fd: i64 = sys_openat_wr(path, BR_FILE_MODE)
215 if fd < 0 { return 0 - 1 }
216 br_write_all(fd, b, o)
217 sys_close(fd)
218 return 0
219}
220// flatten TAB, CR and NL to spaces so the rewrite stays one TSV field
221func br_flatten(s: *u8, n: i64) -> i64 {
222 var i: i64 = 0
223 while i < n {
224 let c: i64 = s[i] as i64
225 if c == BR_TAB { s[i] = BR_SPACE as u8 }
226 if c == BR_NL { s[i] = BR_SPACE as u8 }
227 if c == BR_CR { s[i] = BR_SPACE as u8 }
228 i = i + 1
229 }
230 return n
231}
232
233// ---- the remote door -------------------------------------------------------------------------------------------
234// eng: [0..3]=a.b.c.d [4]=port [5]=reply timeout s [6]=1 when an engine was declared [7]=kind (BR_ENGINE_KIND_*)
235func br_eng_alloc() -> *i64 { let e: *i64 = sys_mmap(BR_ENG_SLOTS * BR_I64) as *i64; e[6] = 0; e[7] = BR_ENGINE_KIND_NISHI; return e }
236// a permil as a JSON decimal ("700" -> 0.700, "1000" -> 1.000): the oracle wire wants floats, this estate carries none
237func br_cat_permil(dst: *u8, off: i64, v: i64) -> i64 {
238 var o: i64 = br_catn(dst, off, v / BR_PERMIL)
239 dst[o] = 46 as u8; o = o + 1
240 let f: i64 = v % BR_PERMIL
241 dst[o] = (48 + f / 100) as u8; o = o + 1
242 dst[o] = (48 + (f / 10) % 10) as u8; o = o + 1
243 dst[o] = (48 + f % 10) as u8; o = o + 1
244 dst[o] = 0 as u8
245 return o
246}
247// "a.b.c.d:port" -> eng; 1 ok / 0 malformed (every octet 0..255, port 1..65535)
248func br_engine_parse(s: *u8, eng: *i64) -> i64 {
249 var i: i64 = 0
250 var part: i64 = 0
251 var v: i64 = 0
252 var digits: i64 = 0
253 var ok: i64 = 1
254 var go: i64 = 1
255 while go == 1 {
256 let c: i64 = s[i] as i64
257 if c == 0 { go = 0 } else {
258 if c >= 48 { if c <= 57 { v = v * 10 + (c - 48); digits = digits + 1 } }
259 if c == 46 { if part > 2 { ok = 0 } if digits == 0 { ok = 0 } if v > 255 { ok = 0 } eng[part] = v; part = part + 1; v = 0; digits = 0 }
260 if c == 58 { if part != 3 { ok = 0 } if digits == 0 { ok = 0 } if v > 255 { ok = 0 } eng[3] = v; part = 4; v = 0; digits = 0 }
261 if c < 46 { ok = 0 }
262 if c > 58 { ok = 0 }
263 if c == 47 { ok = 0 }
264 i = i + 1
265 }
266 if ok == 0 { go = 0 }
267 }
268 if ok == 0 { return 0 }
269 if part != 4 { return 0 }
270 if digits == 0 { return 0 }
271 if v < 1 { return 0 }
272 if v > 65535 { return 0 }
273 eng[4] = v
274 eng[6] = 1
275 return 1
276}
277func br_engine_timeout_s(maxnew: i64) -> i64 { return (maxnew * BR_ENGINE_TOKEN_MS_BOUND) / 1000 + BR_ENGINE_SETUP_S }
278// the /gen request body -- the same contract the serve's own page posts; chat:1 = the ChatML wrap the in-process door uses (gp[12])
279func br_engine_body(dst: *u8, cap: i64, prompt: *u8, pn: i64, maxnew: i64, temp: i64, topp: i64, topk: i64, seed: i64) -> i64 {
280 if pn * 2 + BR_ENGINE_BODY_RESERVE > cap { return 0 - 1 }
281 var o: i64 = nsv_cat(dst, 0, "{\"prompt\":\"" as *u8)
282 o = nsv_jesc(dst, o, prompt, pn)
283 o = nsv_cat(dst, o, "\",\"max_new\":" as *u8); o = nsv_catn(dst, o, maxnew)
284 o = nsv_cat(dst, o, ",\"mode\":\"i8\",\"temp\":" as *u8); o = nsv_catn(dst, o, temp)
285 o = nsv_cat(dst, o, ",\"top_p\":" as *u8); o = nsv_catn(dst, o, topp)
286 o = nsv_cat(dst, o, ",\"top_k\":" as *u8); o = nsv_catn(dst, o, topk)
287 o = nsv_cat(dst, o, ",\"seed\":" as *u8); o = nsv_catn(dst, o, seed)
288 o = nsv_cat(dst, o, ",\"chat\":" as *u8); o = nsv_catn(dst, o, BR_CHATML)
289 o = nsv_cat(dst, o, "}" as *u8)
290 return o
291}
292// the llama.cpp /completion body: the prompt is wrapped in the sovereign ChatML shape HERE (the oracle does not wrap),
293// sampling travels as decimals, and the prompt cache is off so every query is answered from the same cold state
294func br_engine_body_llamacpp(dst: *u8, cap: i64, prompt: *u8, pn: i64, maxnew: i64, temp: i64, topp: i64, topk: i64, seed: i64) -> i64 {
295 if pn * 2 + BR_ENGINE_BODY_RESERVE > cap { return 0 - 1 }
296 var o: i64 = nsv_cat(dst, 0, "{\"prompt\":\"" as *u8)
297 o = nsv_jesc(dst, o, BR_CHATML_HEAD, br_slen(BR_CHATML_HEAD))
298 o = nsv_jesc(dst, o, prompt, pn)
299 o = nsv_jesc(dst, o, BR_CHATML_TAIL, br_slen(BR_CHATML_TAIL))
300 o = nsv_cat(dst, o, "\",\"n_predict\":" as *u8); o = nsv_catn(dst, o, maxnew)
301 o = nsv_cat(dst, o, ",\"temperature\":" as *u8); o = br_cat_permil(dst, o, temp)
302 o = nsv_cat(dst, o, ",\"top_p\":" as *u8); o = br_cat_permil(dst, o, topp)
303 o = nsv_cat(dst, o, ",\"top_k\":" as *u8); o = nsv_catn(dst, o, topk)
304 o = nsv_cat(dst, o, ",\"seed\":" as *u8); o = nsv_catn(dst, o, seed)
305 o = nsv_cat(dst, o, ",\"cache_prompt\":false,\"stream\":false}" as *u8)
306 return o
307}
308// classify ONE raw llama.cpp reply (pure): 200 with "content" -> text; a 400 whose message names the context window is
309// a per-prompt REFUSAL; every other shape ABORTS. meta as for the sovereign parser ([1]=tokens_predicted [2]=predicted_ms)
310func br_engine_parse_resp_llamacpp(resp: *u8, n: i64, outb: *u8, ocap: i64, meta: *i64) -> i64 {
311 meta[0] = 0; meta[1] = 0; meta[2] = 0; meta[3] = 0; meta[4] = 0; meta[5] = 0
312 if n <= 0 { return BR_ENG_ABORT }
313 let pr: *i64 = nx_http_resp_alloc()
314 let pv: i64 = nx_http_response_parse(resp, n, pr)
315 if pv != NX_HTTP_RESP_OK { return BR_ENG_ABORT }
316 meta[3] = pr[1]
317 let bo: i64 = pr[6]
318 var bl: i64 = pr[7]
319 if bl < 0 { bl = n - bo }
320 if bo < 0 { return BR_ENG_ABORT }
321 if bo + bl > n { bl = n - bo }
322 let bb: *u8 = resp + bo
323 if pr[1] == BR_HTTP_BAD {
324 if br_find(bb, bl, "context" as *u8) >= 0 { meta[5] = BR_HTTP_BAD; return BR_ENG_REFUSED }
325 return BR_ENG_ABORT
326 }
327 if pr[1] != BR_HTTP_OK { return BR_ENG_ABORT }
328 let kt: i64 = nsv_jkey(bb, bl, "content" as *u8)
329 if kt < 0 { return BR_ENG_ABORT }
330 let tl: i64 = nsv_jstr(bb, bl, kt, outb, ocap)
331 if tl < 0 { return BR_ENG_ABORT }
332 meta[0] = nsv_jint(bb, bl, nsv_jkey(bb, bl, "tokens_evaluated" as *u8), 0)
333 meta[1] = nsv_jint(bb, bl, nsv_jkey(bb, bl, "tokens_predicted" as *u8), 0)
334 meta[2] = nsv_jint(bb, bl, nsv_jkey(bb, bl, "predicted_ms" as *u8), 0)
335 return tl
336}
337// classify ONE raw HTTP reply (pure). Returns the text length written to outb (>= 0), BR_ENG_REFUSED with meta[5]=err
338// for a 400 the engine explains, or BR_ENG_ABORT for anything the door must not turn into a passthrough row.
339// meta: [0]=prompt_tokens [1]=gen_tokens [2]=ms_total [3]=http status [4]=eos [5]=err
340func br_engine_parse_resp(resp: *u8, n: i64, outb: *u8, ocap: i64, meta: *i64) -> i64 {
341 meta[0] = 0; meta[1] = 0; meta[2] = 0; meta[3] = 0; meta[4] = 0; meta[5] = 0
342 if n <= 0 { return BR_ENG_ABORT }
343 let pr: *i64 = nx_http_resp_alloc()
344 let pv: i64 = nx_http_response_parse(resp, n, pr)
345 if pv != NX_HTTP_RESP_OK { return BR_ENG_ABORT }
346 meta[3] = pr[1]
347 let bo: i64 = pr[6]
348 var bl: i64 = pr[7]
349 if bl < 0 { bl = n - bo }
350 if bo < 0 { return BR_ENG_ABORT }
351 if bo + bl > n { bl = n - bo }
352 let bb: *u8 = resp + bo
353 let ok: i64 = nsv_jint(bb, bl, nsv_jkey(bb, bl, "ok" as *u8), 0 - 1)
354 if pr[1] == BR_HTTP_BAD {
355 let err: i64 = nsv_jint(bb, bl, nsv_jkey(bb, bl, "err" as *u8), 0 - 1)
356 if err < 0 { return BR_ENG_ABORT }
357 if err == BR_ENGINE_ERR_MALFORMED { return BR_ENG_ABORT }
358 meta[5] = err
359 return BR_ENG_REFUSED
360 }
361 if pr[1] != BR_HTTP_OK { return BR_ENG_ABORT }
362 if ok != 1 { return BR_ENG_ABORT }
363 let kt: i64 = nsv_jkey(bb, bl, "text" as *u8)
364 if kt < 0 { return BR_ENG_ABORT }
365 let tl: i64 = nsv_jstr(bb, bl, kt, outb, ocap)
366 if tl < 0 { return BR_ENG_ABORT }
367 meta[0] = nsv_jint(bb, bl, nsv_jkey(bb, bl, "prompt_tokens" as *u8), 0)
368 meta[1] = nsv_jint(bb, bl, nsv_jkey(bb, bl, "gen_tokens" as *u8), 0)
369 meta[2] = nsv_jint(bb, bl, nsv_jkey(bb, bl, "ms_total" as *u8), 0)
370 meta[4] = nsv_jint(bb, bl, nsv_jkey(bb, bl, "eos" as *u8), 0)
371 return tl
372}
373// one prompt through the LAN door. Returns the text length, or BR_ENG_UNREACHABLE / BR_ENG_ABORT / BR_ENG_REFUSED.
374func br_engine_gen(eng: *i64, prompt: *u8, pn: i64, maxnew: i64, temp: i64, topp: i64, topk: i64, seed: i64, outb: *u8, ocap: i64, meta: *i64) -> i64 {
375 meta[0] = 0; meta[1] = 0; meta[2] = 0; meta[3] = 0; meta[4] = 0; meta[5] = 0
376 let bcap: i64 = pn * 2 + BR_ENGINE_BODY_RESERVE
377 let body: *u8 = sys_mmap(bcap)
378 var bn: i64 = 0
379 var gpath: *u8 = BR_GEN_PATH
380 if eng[7] == BR_ENGINE_KIND_LLAMACPP {
381 bn = br_engine_body_llamacpp(body, bcap, prompt, pn, maxnew, temp, topp, topk, seed)
382 gpath = BR_GEN_PATH_LLAMACPP
383 } else {
384 bn = br_engine_body(body, bcap, prompt, pn, maxnew, temp, topp, topk, seed)
385 }
386 if bn < 0 { return BR_ENG_ABORT }
387 let host: *u8 = sys_mmap(BR_ENGINE_HOST_CAP)
388 var ho: i64 = br_catn(host, 0, eng[0]); host[ho] = 46 as u8; ho = ho + 1
389 ho = br_catn(host, ho, eng[1]); host[ho] = 46 as u8; ho = ho + 1
390 ho = br_catn(host, ho, eng[2]); host[ho] = 46 as u8; ho = ho + 1
391 ho = br_catn(host, ho, eng[3]); host[ho] = 58 as u8; ho = ho + 1
392 ho = br_catn(host, ho, eng[4])
393 let plen: i64 = br_slen(gpath)
394 let rcap: i64 = nx_http_client_request_cap(plen, ho, 0, 0) + bn
395 let req: *u8 = sys_mmap(rcap)
396 let rn: i64 = nx_http_client_build_request_post(gpath, plen, host, ho, BR_GEN_CTYPE, br_slen(BR_GEN_CTYPE), body, bn, req)
397 let addr: *u8 = sys_mmap(16)
398 nx_http_client_sockaddr_ipv4(addr, eng[0], eng[1], eng[2], eng[3], eng[4])
399 let fd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
400 if fd < 0 { return BR_ENG_UNREACHABLE }
401 sys_set_socket_timeout(fd, eng[5])
402 if sys_connect(fd, addr, 16) < 0 { sys_close(fd); return BR_ENG_UNREACHABLE }
403 if br_write_all(fd, req, rn) != 0 { sys_close(fd); return BR_ENG_UNREACHABLE }
404 let resp: *u8 = sys_mmap(BR_ENGINE_RESP_CAP)
405 var got: i64 = 0
406 var keep: i64 = 1
407 while keep == 1 {
408 if got >= BR_ENGINE_RESP_CAP - 1 { keep = 0 } else {
409 let r: i64 = sys_read(fd, resp + got, BR_ENGINE_RESP_CAP - 1 - got)
410 if r <= 0 { keep = 0 } else { got = got + r }
411 }
412 }
413 sys_close(fd)
414 if got <= 0 { return BR_ENG_UNREACHABLE }
415 if eng[7] == BR_ENGINE_KIND_LLAMACPP { return br_engine_parse_resp_llamacpp(resp, got, outb, ocap, meta) }
416 return br_engine_parse_resp(resp, got, outb, ocap, meta)
417}
418
419// ---- the split verdict (pure) -----------------------------------------------------------------------------------
420// rows = queries in the split, done = have + written, written = rows written THIS run, pass = passthrough rows
421// written this run. A split is COMPLETE only when every query has a row AND this run produced at least one real
422// rewrite (or wrote nothing new, i.e. it merely confirmed a full part file).
423func br_split_verdict(rows: i64, done: i64, written: i64, pass: i64, limit: i64) -> i64 {
424 if limit > 0 { return BR_STATE_SMOKE }
425 if done < rows { return BR_STATE_PARTIAL }
426 if rows == 0 { return BR_STATE_PARTIAL }
427 if written > 0 { if pass >= written { return BR_STATE_ALL_PASSTHROUGH } }
428 return BR_STATE_COMPLETE
429}
430func br_state_name(s: i64) -> *u8 {
431 if s == BR_STATE_COMPLETE { return "COMPLETE" as *u8 }
432 if s == BR_STATE_SMOKE { return "SMOKE" as *u8 }
433 if s == BR_STATE_EXISTING { return "COMPLETE-EXISTING" as *u8 }
434 if s == BR_STATE_LEASED { return "LEASED-ELSEWHERE" as *u8 }
435 if s == BR_STATE_ENGINE_DOWN { return "ENGINE-DOWN" as *u8 }
436 if s == BR_STATE_ALL_PASSTHROUGH { return "ALL-PASSTHROUGH" as *u8 }
437 if s == BR_STATE_LEASE_IO { return "LEASE-IO" as *u8 }
438 return "PARTIAL" as *u8
439}
440func br_lease_name(dst: *u8, out: *u8, split: *u8) -> i64 {
441 var o: i64 = br_cat(dst, 0, BR_LEASE_NAME_PFX)
442 o = br_cat(dst, o, out)
443 dst[o] = 45 as u8; o = o + 1
444 o = br_cat(dst, o, split)
445 return o
446}
447func br_owner(dst: *u8) -> i64 { var o: i64 = br_cat(dst, 0, "pid" as *u8); o = br_catn(dst, o, ls_pid()); return o }
448
449// ---- one split ---------------------------------------------------------------------------------------------------
450// cfg: [0]=concat [1]=maxnew [2]=limit [3]=temp [4]=topp [5]=topk [6]=seed [7]=lease_ttl
451// st: [0]=rows_total [1]=rows_written_now [2]=passthrough [3]=skipped_existing [4]=n_gen_sum [5]=ms_sum
452// [6]=complete(1) [7]=smoke(1) [8]=state [9]=engine_rc_on_abort [10]=holder_known(1)
453// returns 0 ran (read st) / -1 split unreadable or unwritable
454func br_run_split(root: *u8, split: *u8, out: *u8, cfg: *i64, eng: *i64, progress: *u8, st: *i64) -> i64 {
455 var z: i64 = 0
456 while z < BR_ST_SLOTS { st[z] = 0; z = z + 1 }
457 let concat: i64 = cfg[0]
458 let maxnew: i64 = cfg[1]
459 let limit: i64 = cfg[2]
460 let temp: i64 = cfg[3]
461 let topp: i64 = cfg[4]
462 let topk: i64 = cfg[5]
463 let seed: i64 = cfg[6]
464 let tail: *u8 = sys_mmap(BR_PATH_CAP)
465 var to: i64 = br_cat(tail, 0, "queries." as *u8); to = br_cat(tail, to, out); to = br_cat(tail, to, ".tsv" as *u8)
466 let fpath: *u8 = sys_mmap(BR_PATH_CAP)
467 br_path(fpath, root, split, tail)
468 // idempotent re-run: a finished split is skipped without touching the engine
469 let ffd: i64 = sys_openat_rd(fpath)
470 if ffd >= 0 {
471 sys_close(ffd)
472 st[8] = BR_STATE_EXISTING
473 br_puts(" " as *u8); br_puts(split); br_puts(": state=COMPLETE-EXISTING -> " as *u8); br_puts(fpath); br_puts("\n" as *u8)
474 return 0
475 }
476 // the lease: two runners on one tree take disjoint splits
477 let lname: *u8 = sys_mmap(BR_PATH_CAP)
478 br_lease_name(lname, out, split)
479 let owner: *u8 = sys_mmap(BR_NUM_CAP * 2)
480 br_owner(owner)
481 let hb: *u8 = sys_mmap(BR_PATH_CAP)
482 // cfg[8] may carry a lease-root pointer (a gate keeps its leases under /tmp/<gate>/); 0 = the estate's default root
483 var lroot: *u8 = LS_DEFAULT_ROOT
484 if cfg[8] != 0 { lroot = cfg[8] as *u8 } else { ls_mkdir("knowledge" as *u8) }
485 let lr: i64 = ls_acquire_root(lroot, lname, owner, cfg[7], hb)
486 if lr == 1 {
487 st[8] = BR_STATE_LEASED
488 st[10] = 1
489 br_puts(" " as *u8); br_puts(split); br_puts(": state=LEASED-ELSEWHERE holder=" as *u8); br_puts(hb); br_puts(" lease=" as *u8); br_puts(lname); br_puts("\n" as *u8)
490 return 0
491 }
492 if lr < 0 {
493 st[8] = BR_STATE_LEASE_IO
494 br_puts(" " as *u8); br_puts(split); br_puts(": state=LEASE-IO lease=" as *u8); br_puts(lname); br_puts(" (fail closed: no split runs without its lease)\n" as *u8)
495 return 0
496 }
497 let qpath: *u8 = sys_mmap(BR_PATH_CAP)
498 br_path(qpath, root, split, "queries.tsv" as *u8)
499 let qn: *i64 = sys_mmap(BR_I64 * 2) as *i64
500 let q: *u8 = sys_read_file(qpath, qn)
501 if (q as i64) == 0 { br_puts(" " as *u8); br_puts(split); br_puts(": UNREADABLE " as *u8); br_puts(qpath); br_puts("\n" as *u8); ls_release_root(lroot, lname, owner); return 0 - 1 }
502 let n: i64 = qn[0]
503 st[0] = br_count_lines(q, n)
504 if n > 0 { if q[n - 1] != (BR_NL as u8) { st[0] = st[0] + 1 } }
505 to = br_cat(tail, to, ".part" as *u8)
506 let ppath: *u8 = sys_mmap(BR_PATH_CAP)
507 br_path(ppath, root, split, tail)
508 // resume: rows already in the part file are skipped by count (rows are appended in query order)
509 let pn: *i64 = sys_mmap(BR_I64 * 2) as *i64
510 let pb: *u8 = sys_read_file(ppath, pn)
511 var have: i64 = 0
512 if (pb as i64) != 0 { have = br_count_lines(pb, pn[0]) }
513 st[3] = have
514 let fd: i64 = __syscall(BR_SYS_OPENAT, BR_AT_FDCWD, ppath, BR_O_APPEND_CREATE, BR_FILE_MODE, 0, 0)
515 if fd < 0 { br_puts(" " as *u8); br_puts(split); br_puts(": CANNOT-OPEN " as *u8); br_puts(ppath); br_puts("\n" as *u8); ls_release_root(lroot, lname, owner); return 0 - 1 }
516 let gp: *i64 = sys_mmap(BR_GP_SLOTS * BR_I64) as *i64
517 let meta: *i64 = sys_mmap(BR_META_SLOTS * BR_I64) as *i64
518 let outb: *u8 = sys_mmap(BR_OUT_CAP)
519 let headn: i64 = br_slen(BR_PROMPT_HEAD)
520 let tailn: i64 = br_slen(BR_PROMPT_TAIL)
521 var idx: i64 = 0
522 var p: i64 = 0
523 var written: i64 = 0
524 var stopped: i64 = 0
525 var engine_down: i64 = 0
526 while p < n {
527 if stopped == 1 { p = n } else {
528 // line end: the first NL at or after p, or n when the last line has none
529 var lnl: i64 = n
530 var j: i64 = p
531 var found: i64 = 0
532 while j < n { if found == 0 { if q[j] == (BR_NL as u8) { lnl = j; found = 1 } } j = j + 1 }
533 var tend: i64 = lnl
534 if tend > p { if q[tend - 1] == (BR_CR as u8) { tend = tend - 1 } }
535 if tend > p {
536 if idx < have { st[3] = st[3] } else {
537 if limit > 0 { if written >= limit { stopped = 1 } }
538 if stopped == 0 {
539 var tb: i64 = p
540 while tb < tend { if q[tb] == (BR_TAB as u8) { j = tb; tb = tend } else { tb = tb + 1 } }
541 var idlen: i64 = tend - p
542 var txt: i64 = tend
543 if j >= p { if j < tend { if q[j] == (BR_TAB as u8) { idlen = j - p; txt = j + 1 } } }
544 let tlen: i64 = tend - txt
545 let prompt: *u8 = sys_mmap(headn + tlen + tailn + 1)
546 var po: i64 = br_cat(prompt, 0, BR_PROMPT_HEAD)
547 po = br_catb(prompt, po, q + txt, tlen)
548 po = br_cat(prompt, po, BR_PROMPT_TAIL)
549 let t0: i64 = sys_now_ms()
550 var tl: i64 = 0
551 if eng[6] == 1 {
552 tl = br_engine_gen(eng, prompt, po, maxnew, temp, topp, topk, seed, outb, BR_OUT_USE, meta)
553 } else {
554 gp[0] = prompt as i64; gp[1] = po; gp[2] = maxnew; gp[3] = BR_MODE_I8
555 gp[4] = outb as i64; gp[5] = BR_OUT_USE; gp[6] = meta as i64; gp[7] = BR_NOFD
556 gp[8] = temp; gp[9] = topp; gp[10] = topk; gp[11] = seed
557 gp[12] = BR_CHATML; gp[13] = BR_NOSTREAM
558 tl = nsv_generate(gp)
559 if tl <= 0 { tl = BR_ENG_REFUSED }
560 if meta[5] != 0 { tl = BR_ENG_REFUSED }
561 }
562 let ms: i64 = sys_now_ms() - t0
563 if tl == BR_ENG_UNREACHABLE { engine_down = 1 }
564 if tl == BR_ENG_ABORT { engine_down = 1 }
565 if engine_down == 1 {
566 st[9] = tl
567 stopped = 1
568 br_puts(" " as *u8); br_puts(split); br_puts(": ENGINE-DOWN rc=" as *u8); br_putn(tl); br_puts(" http=" as *u8); br_putn(meta[3]); br_puts(" after ms=" as *u8); br_putn(ms); br_puts(" at query " as *u8); br_putn(have + written + 1); br_puts(" -- the row is NOT written; the split resumes on the next run\n" as *u8)
569 } else {
570 var pass: i64 = 0
571 if tl < 0 { pass = 1 }
572 if pass == 1 { st[2] = st[2] + 1 } else { st[4] = st[4] + meta[1] }
573 st[5] = st[5] + ms
574 // the row: id TAB (original SPACE)? rewrite NL -- the rewrite flattened to one field
575 var rl: i64 = tl
576 if pass == 1 { rl = 0 } else { br_flatten(outb, tl) }
577 let row: *u8 = sys_mmap(idlen + 1 + tlen + 1 + rl + 2)
578 var ro: i64 = br_catb(row, 0, q + p, idlen)
579 row[ro] = BR_TAB as u8; ro = ro + 1
580 if pass == 1 { ro = br_catb(row, ro, q + txt, tlen) } else {
581 if concat == 1 { ro = br_catb(row, ro, q + txt, tlen); row[ro] = BR_SPACE as u8; ro = ro + 1 }
582 ro = br_catb(row, ro, outb, rl)
583 }
584 row[ro] = BR_NL as u8; ro = ro + 1
585 if br_write_all(fd, row, ro) != 0 { br_puts(" " as *u8); br_puts(split); br_puts(": WRITE-FAILED\n" as *u8); sys_close(fd); ls_release_root(lroot, lname, owner); return 0 - 1 }
586 written = written + 1
587 st[1] = written
588 br_progress(progress, split, have + written, st[0], meta[1], ms, st[2])
589 }
590 }
591 }
592 idx = idx + 1
593 }
594 p = lnl + 1
595 }
596 }
597 sys_close(fd)
598 var state: i64 = br_split_verdict(st[0], have + written, written, st[2], limit)
599 if engine_down == 1 { state = BR_STATE_ENGINE_DOWN }
600 st[8] = state
601 if state == BR_STATE_SMOKE { st[7] = 1 }
602 if state == BR_STATE_COMPLETE { st[6] = 1; sys_renameat(ppath, fpath) }
603 ls_release_root(lroot, lname, owner)
604 br_puts(" " as *u8); br_puts(split); br_puts(": rows=" as *u8); br_putn(st[0])
605 br_puts(" existing=" as *u8); br_putn(have); br_puts(" written=" as *u8); br_putn(written)
606 br_puts(" passthrough=" as *u8); br_putn(st[2]); br_puts(" n_gen=" as *u8); br_putn(st[4]); br_puts(" ms=" as *u8); br_putn(st[5])
607 br_puts(" state=" as *u8); br_puts(br_state_name(state))
608 if state == BR_STATE_COMPLETE { br_puts(" -> " as *u8); br_puts(fpath) }
609 if state == BR_STATE_ALL_PASSTHROUGH { br_puts(" (every row this run was the original query: the part file is kept and NOT renamed -- an engine that refuses everything has produced nothing)" as *u8) }
610 br_puts("\n" as *u8)
611 return 0
612}