nx_gen_worker.nx source

1// nx_gen_worker.nx -- the GenWorker object: the GPU-worker CLIENT. Owns dispatch to a worker (a.b.c.d:port) for 2// image gen (/v1/images/generations) and chat (/v1/chat/completions), plus the non-blocking TCP health-check that 3// makes multi-worker availability-based selection + failover safe (a powered-off worker must never hang a dispatch). 4// Extracted from nx_gen_orchestrator (monolith -> objects); behaviour-preserving. THIS is where ER-8 model-select-by- 5// intent will land (route SFW->controlled model, explicit->turbo). Self-contained (private gw_* helpers). Object 6// model: companion_arch.md. license_tier: ORIGINAL 7import "nx_syscalls.nx" 8import "nx_netprobe_lib.nx" 9import "nx_gen_conf_lib.nx" // G7/G14 (2026-08-30): the lane's one two-path conf reader 10 11func gw_cat(dst: *u8, off: i64, s: *u8) -> i64 { var o: i64=off; var i: i64=0; while s[i]!=(0 as u8){dst[o]=s[i]; o=o+1; i=i+1} return o } 12func gw_catb(dst: *u8, off: i64, src: *u8, n: i64) -> i64 { var o: i64=off; var i: i64=0; while i<n {dst[o]=src[i]; o=o+1; i=i+1} return o } 13func gw_itoa(dst: *u8, off: i64, v: i64) -> i64 { let t: *u8=sys_mmap(28); var m: i64=v; 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 o: i64=off; var q: i64=k-1; while q>=0{dst[o]=t[q];o=o+1;q=q-1} return o } 14 15// ONE read loop for every dispatch in this object. A read ERROR (r<0 -- which SO_RCVTIMEO delivers as 16// EAGAIN once the socket timeout expires) is NOT a clean EOF (r==0). The previous `if r <= 0` collapsed 17// them onto one branch, so the function returned 0 for BOTH, every caller tested only >0, the candidate 18// loop exhausted, and the daemon reported worker ABSENCE for a worker that was connected and rendering. 19// MEASURED 2026-08-29: a socket watch showed 192.168.8.192:7861 <- 192.168.8.227:51938 ESTABLISHED and 20// held for 164s while the caller was being told no gpu worker was available. The return space was already 21// three-valued (0-1 socket, 0-2 connect) and the caller tested one boolean. 22// >0 bytes read 23// 0 clean EOF with nothing read (peer closed without answering) 24// 0-3 TIMEOUT with nothing read -- connected and busy; the render may still be completing on the GPU 25// 0-4 PARTIAL read then timeout -- the response is truncated, which previously returned a POSITIVE 26// byte count and read as SUCCESS, resurfacing later as "worker returned no images" 27// EXTRACTED, not re-typed: the identical loop existed in go_worker_dispatch, go_worker_img2img and 28// go_chat_dispatch. A fix applied to one sibling and not the others is half a fix, and the half left 29// undone is the one that ships. 30func gw_read_full(fd: i64, respbuf: *u8, respcap: i64) -> i64 { 31 var total: i64 = 0 32 var gof: i64 = 1 33 var lastr: i64 = 0 34 while gof == 1 { 35 let r: i64 = sys_read(fd, (respbuf as i64 + total) as *u8, respcap - total) 36 if r < 0 { lastr = r; gof = 0 } else { if r == 0 { gof = 0 } else { total = total + r; if total >= respcap { gof = 0 } } } 37 } 38 if lastr < 0 { if total == 0 { return 0 - 3 } else { return 0 - 4 } } 39 return total 40} 41 42// dispatch an image-gen request: POST /v1/images/generations to worker (a.b.c.d:port), read full response. 43func go_worker_dispatch(ha: i64, hb: i64, hc: i64, hd: i64, port: i64, body: *u8, blen: i64, respbuf: *u8, respcap: i64) -> i64 { 44 let fd: i64 = sys_socket(2, 1, 0) 45 if fd < 0 { return 0 - 1 } 46 sys_set_socket_timeout(fd, 180) 47 let a: *u8 = sys_mmap(16) 48 a[0]=2 as u8; a[1]=0 as u8; a[2]=((port>>8)&0xff) as u8; a[3]=(port&0xff) as u8 49 a[4]=ha as u8; a[5]=hb as u8; a[6]=hc as u8; a[7]=hd as u8 50 var zi: i64=8; while zi<16 { a[zi]=0 as u8; zi=zi+1 } 51 if sys_connect(fd, a, 16) != 0 { sys_close(fd); return 0 - 2 } 52 let rq: *u8 = sys_mmap(16384); var o: i64 = 0 53 o = gw_cat(rq, o, "POST /v1/images/generations HTTP/1.1\r\nHost: 127.0.0.1\r\nAccept: */*\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: " as *u8) 54 o = gw_itoa(rq, o, blen); o = gw_cat(rq, o, "\r\n\r\n" as *u8); o = gw_catb(rq, o, body, blen) 55 sys_write(fd, rq, o) 56 let total: i64 = gw_read_full(fd, respbuf, respcap) 57 sys_close(fd) 58 return total 59} 60// dispatch an IMG2IMG request: POST /sdapi/v1/img2img (A1111 shape, which stable-diffusion.cpp serves) to the worker. 61// The body carries a base64 init image (~1.4MB+ for 512px) so we write a SMALL header then STREAM the caller's body 62// from its pointer in a write-loop (no multi-MB request copy, and one sys_write cannot flush a MB in a single call). 63// Response is {"images":["<b64>"]} -> the SAME gp_response A1111 branch the txt2img path already ingests. This is 64// where structure-preserving edits (digital-twin refine, attribute change, and mask-based inpaint) dispatch. 65func go_worker_img2img(ha: i64, hb: i64, hc: i64, hd: i64, port: i64, body: *u8, blen: i64, respbuf: *u8, respcap: i64) -> i64 { 66 let fd: i64 = sys_socket(2, 1, 0) 67 if fd < 0 { return 0 - 1 } 68 sys_set_socket_timeout(fd, 200) 69 let a: *u8 = sys_mmap(16) 70 a[0]=2 as u8; a[1]=0 as u8; a[2]=((port>>8)&0xff) as u8; a[3]=(port&0xff) as u8 71 a[4]=ha as u8; a[5]=hb as u8; a[6]=hc as u8; a[7]=hd as u8 72 var zi: i64=8; while zi<16 { a[zi]=0 as u8; zi=zi+1 } 73 if sys_connect(fd, a, 16) != 0 { sys_close(fd); return 0 - 2 } 74 let hdr: *u8 = sys_mmap(1024); var ho: i64 = 0 75 ho = gw_cat(hdr, 0, "POST /sdapi/v1/img2img HTTP/1.1\r\nHost: 127.0.0.1\r\nAccept: */*\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: " as *u8) 76 ho = gw_itoa(hdr, ho, blen); ho = gw_cat(hdr, ho, "\r\n\r\n" as *u8) 77 sys_write(fd, hdr, ho) 78 var sent: i64 = 0 79 while sent < blen { let w: i64 = sys_write(fd, (body as i64 + sent) as *u8, blen - sent); if w <= 0 { sent = blen } else { sent = sent + w } } 80 let total: i64 = gw_read_full(fd, respbuf, respcap) 81 sys_close(fd) 82 return total 83} 84// non-blocking TCP health-check ("electronic availability"): connect non-blocking + poll(POLLOUT, 2s) + SO_ERROR. 85// 1=reachable, 0=down/refused/timeout. = the multi-worker availability selector's liveness probe. 86// extracted to nx_netprobe_lib (np_probe_up) 2026-08-12 -- this was the ORIGINAL copy; nx_swarm_heal's 87// go_probe_up was the duplicate. Now BOTH import the one lib (rule 15). Thin alias keeps every call site 88// (go_worker_up) untouched = behaviour-preserving; the byte-identical probe logic now lives in one place. 89func go_worker_up(ha: i64, hb: i64, hc: i64, hd: i64, port: i64) -> i64 { return np_probe_up(ha, hb, hc, hd, port) } 90// CHAT DISPATCH BUDGET AS DATA (2026-08-30): the per-try read timeout was a literal 25 s; a measured warm completion on 91// the 5080 with a game on the card took 17.6 s, and the daemon's composed system prompt makes real turns longer, so the 92// literal was the ceiling every busy turn hit. knowledge/gen_chat_budget.conf (first integer, seconds) is read per call from 93// the daemon cwd then the nishihost tree; absent, unreadable or zero -> the historical 25, never a silent zero; a ceiling 94// bounds a typo. Same idiom as go_default_steps. 95const GW_CHAT_BUDGET_FALLBACK_S: i64 = 25 96const GW_CHAT_BUDGET_CEIL_S: i64 = 300 97func gw_chat_budget_s() -> i64 { 98 let szp: *i64 = sys_mmap(16) as *i64 99 szp[0] = 0 100 var b: *u8 = sys_read_file("knowledge/gen_chat_budget.conf" as *u8, szp) 101 if szp[0] <= 0 { b = sys_read_file("/volume1/homes/elderwesto/nishihost/knowledge/gen_chat_budget.conf" as *u8, szp) } 102 if szp[0] <= 0 { return GW_CHAT_BUDGET_FALLBACK_S } 103 var v: i64 = 0 104 var seen: i64 = 0 105 var i: i64 = 0 106 while i < szp[0] { 107 let c: i64 = b[i] & 0xff 108 if c >= 48 { if c <= 57 { v = v*10 + (c - 48); seen = 1; i = i + 1 } else { i = szp[0] } } else { if seen == 1 { i = szp[0] } else { i = i + 1 } } 109 } 110 if seen == 0 { return GW_CHAT_BUDGET_FALLBACK_S } 111 if v <= 0 { return GW_CHAT_BUDGET_FALLBACK_S } 112 if v > GW_CHAT_BUDGET_CEIL_S { return GW_CHAT_BUDGET_CEIL_S } 113 return v 114} 115// ---- G7 (2026-08-30) MASK INPAINT: the NishiLang verb over an engine surface that ALREADY answers ---------------- 116// The served sd-server's /sdapi/v1/img2img decodes a base64 PNG "mask" (ONE channel: 255 = repaint, 0 = keep) and 117// honours "inpainting_mask_invert" (elder-sdcpp examples/server/main.cpp, sdapi_any2img img2img branch, read 118// 2026-08-30); with no mask it repaints the WHOLE canvas, which is the plain img2img go_handle_img2img already 119// drives. This verb APPENDS the two mask fields to an img2img JSON body the caller is composing (it emits its own 120// leading comma, so call it after an existing field) and REFUSES by named code instead of emitting a body the 121// engine would either reject or -- the silent class -- decode to an empty mask and render UNMASKED: 122// >= 0 the new offset; the body now carries "mask" and "inpainting_mask_invert" 123// GW_MASK_EMPTY no mask bytes -- a caller that meant plain img2img must not call this at all 124// GW_MASK_NOT_B64 a byte outside A-Z a-z 0-9 + / = (a data: URI prefix, whitespace, or a raw PNG byte) 125// GW_MASK_BAD_INVERT invert is neither 0 nor 1 126// The mask rides in the daemon's request window (go_serve_conn reads 131,071 B), a PRE-EXISTING bound named here 127// rather than widened here: a one-channel PNG mask of a 768x1024 photo is a few KB of base64. 128const GW_MASK_EMPTY: i64 = 0 - 1 129const GW_MASK_NOT_B64: i64 = 0 - 2 130const GW_MASK_BAD_INVERT: i64 = 0 - 3 131const GW_CH_UP_A: i64 = 65 132const GW_CH_UP_Z: i64 = 90 133const GW_CH_LO_A: i64 = 97 134const GW_CH_LO_Z: i64 = 122 135const GW_CH_D0: i64 = 48 136const GW_CH_D9: i64 = 57 137const GW_CH_PLUS: i64 = 43 138const GW_CH_SLASH: i64 = 47 139const GW_CH_EQ: i64 = 61 140func gw_b64_byte_ok(c: i64) -> i64 { 141 if c >= GW_CH_UP_A { if c <= GW_CH_UP_Z { return 1 } } 142 if c >= GW_CH_LO_A { if c <= GW_CH_LO_Z { return 1 } } 143 if c >= GW_CH_D0 { if c <= GW_CH_D9 { return 1 } } 144 if c == GW_CH_PLUS { return 1 } 145 if c == GW_CH_SLASH { return 1 } 146 if c == GW_CH_EQ { return 1 } 147 return 0 148} 149func gw_inpaint_mask(dst: *u8, off: i64, mask_b64: *u8, mlen: i64, invert: i64) -> i64 { 150 if mlen <= 0 { return GW_MASK_EMPTY } 151 if invert != 0 { if invert != 1 { return GW_MASK_BAD_INVERT } } 152 var i: i64 = 0 153 while i < mlen { if gw_b64_byte_ok(mask_b64[i] & 0xff) == 0 { return GW_MASK_NOT_B64 } i = i + 1 } 154 var o: i64 = gw_cat(dst, off, ",\"mask\":\"" as *u8) 155 o = gw_catb(dst, o, mask_b64, mlen) 156 o = gw_cat(dst, o, "\",\"inpainting_mask_invert\":" as *u8) 157 o = gw_itoa(dst, o, invert) 158 return o 159} 160// dispatch a chat request: POST /v1/chat/completions to the worker (same engine; Qwen text-encoder doubles as the 161// companion LLM). Thin authed proxy; the caller owns the messages array. 162func go_chat_dispatch(ha: i64, hb: i64, hc: i64, hd: i64, port: i64, body: *u8, blen: i64, respbuf: *u8, respcap: i64) -> i64 { 163 let fd: i64 = sys_socket(2, 1, 0) 164 if fd < 0 { return 0 - 1 } 165 sys_set_socket_timeout(fd, gw_chat_budget_s()) 166 let a: *u8 = sys_mmap(16) 167 a[0]=2 as u8; a[1]=0 as u8; a[2]=((port>>8)&0xff) as u8; a[3]=(port&0xff) as u8 168 a[4]=ha as u8; a[5]=hb as u8; a[6]=hc as u8; a[7]=hd as u8 169 var zi: i64=8; while zi<16 { a[zi]=0 as u8; zi=zi+1 } 170 if sys_connect(fd, a, 16) != 0 { sys_close(fd); return 0 - 2 } 171 let rq: *u8 = sys_mmap(131072); var o: i64 = 0 172 o = gw_cat(rq, o, "POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nAccept: */*\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: " as *u8) 173 o = gw_itoa(rq, o, blen); o = gw_cat(rq, o, "\r\n\r\n" as *u8); o = gw_catb(rq, o, body, blen) 174 sys_write(fd, rq, o) 175 let total: i64 = gw_read_full(fd, respbuf, respcap) 176 sys_close(fd) 177 return total 178}