nx_tools_timing_serve_gate.nx source
↩ module page · 261 lines · 15149 B
1// Source-linked staged serve_one integration. No daemon startup/seed. Private pinned fixture only.
2import "nx_tools_api_serve.nx"
3import "nx_connect.nx"
4import "nx_gate_verdict.nx"
5const TG_PRIVATE: i64 = 448
6const TG_FILE: i64 = 384
7const TG_EXEC: i64 = 448
8const TG_I64_MAX: i64 = 9223372036854775807
9const TG_MS_PER_SEC: i64 = 1000
10const TG_SOL_SOCKET: i64 = 1
11const TG_SO_LINGER: i64 = 13
12const TG_LINGER_BYTES: i64 = 8
13const TG_WAIT_NOHANG: i64 = 1
14const TG_SIGKILL: i64 = 9
15const TG_PORT_MAX: i64 = 65535
16const TG_SECRET: *u8 = "timing-serve-gate-only-key-not-a-production-capability" as *u8
17static tg_timeout: i64
18static tg_step: i64
19
20func tg_num(s: *u8) -> i64 {
21 var n: i64 = 0; var i: i64 = 0
22 while s[i] != (0 as u8) {
23 let c: i64 = s[i] as i64
24 if c < 48 { return 0-1 }; if c > 57 { return 0-1 }
25 if n > (TG_I64_MAX-(c-48))/10 { return 0-1 }
26 n=n*10+c-48; i=i+1
27 }
28 if i == 0 { return 0-1 }; return n
29}
30func tg_put(path: *u8, b: *u8, n: i64, mode: i64) -> i64 {
31 let fd: i64 = sys_openat_exclusive(path, mode)
32 if fd < 0 { return fd }
33 let r: i64 = ta_journal_write_fd(fd,b,n)
34 let c: i64 = sys_close(fd)
35 if r < 0 { return r }; return c
36}
37func tg_has(b: *u8, n: i64, s: *u8) -> i64 { if (b as i64) <= 0 { return 0 }; if ta_indexof(b,n,s) >= 0 { return 1 }; return 0 }
38func tg_rowhas(b: *u8, n: i64, tool: *u8, phase: *u8, token: *u8) -> i64 {
39 var a: i64=0; var e: i64=0
40 while e <= n {
41 var end: i64=0
42 if e == n { end=1 } else { if b[e] == (10 as u8) { end=1 } }
43 if end == 1 {
44 let row: *u8=((b as i64)+a) as *u8
45 if tg_has(row,e-a,tool) == 1 { if tg_has(row,e-a,phase) == 1 { if tg_has(row,e-a,token) == 1 { return 1 } } }
46 a=e+1
47 }
48 e=e+1
49 }; return 0
50}
51func tg_wait(pid: i64) -> i64 {
52 let st: *i64=sys_mmap(8) as *i64
53 let start: i64=sys_now_ms()
54 while sys_now_ms()-start < tg_timeout {
55 let r: i64=sys_wait4(pid,st,TG_WAIT_NOHANG)
56 if r == pid { return st[0] }
57 if r < 0 { return 0-1 }
58 sys_sleep_ms(tg_step)
59 }
60 nx_kill(pid,TG_SIGKILL); sys_wait4(pid,st,0); return 0-1
61}
62func tg_case(lfd: i64, addr: *u8, tool: *u8, async: i64, disconnected: i64, phase: *u8, ctr: *i64) -> i64 {
63 let tok: *u8=sys_mmap(TA_MAGIC_4096)
64 let ttl: i64=tg_timeout/TG_MS_PER_SEC+1
65 let tn: i64=capt_issue(TG_SECRET,ta_slen(TG_SECRET),tool,ta_slen(tool),sys_now_realtime_sec()+ttl,1,tok,TA_MAGIC_4096)
66 if tn <= 0 { gv_check("fixture capability issued" as *u8,0,ctr); return 0 }
67 let body: *u8=sys_mmap(TA_MAGIC_4096)
68 var bn: i64=ta_cat(body,0,"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"" as *u8)
69 bn=ta_cat(body,bn,tool); bn=ta_cat(body,bn,"\",\"_cap\":\"" as *u8); bn=ta_catb(body,bn,tok,tn)
70 bn=ta_cat(body,bn,"\"" as *u8)
71 if async == 1 { bn=ta_cat(body,bn,",\"_async\":\"1\"" as *u8) }
72 bn=ta_cat(body,bn,"}}" as *u8)
73 let req: *u8=sys_mmap(TA_MAGIC_4096)
74 var rn: i64=ta_cat(req,0,"POST /mcp HTTP/1.1\r\nHost: gate.invalid\r\nContent-Type: application/json\r\nContent-Length: " as *u8)
75 rn=ta_catn(req,rn,bn); rn=ta_cat(req,rn,"\r\nConnection: close\r\n\r\n" as *u8); rn=ta_catb(req,rn,body,bn)
76 let client: i64=sys_socket(2,1,0)
77 if client < 0 { gv_check("client socket" as *u8,0,ctr); return 0 }
78 if nx_connect_bounded(client,addr,16,tg_timeout) < 0 { sys_close(client); gv_check("loopback connect" as *u8,0,ctr); return 0 }
79 let verdict: *i64=sys_mmap(8) as *i64
80 let server: i64=nx_http_server_accept_one(lfd,verdict)
81 if server < 0 { sys_close(client); gv_check("accept" as *u8,0,ctr); return 0 }
82 sys_set_socket_timeout(client,ttl); sys_set_socket_timeout(server,ttl)
83 let pid: i64=sys_fork()
84 if pid == 0 {
85 sys_close(client); sys_close(lfd)
86 let input: *u8=sys_mmap(EDGE_REQ_MAX)
87 let output: *u8=sys_mmap(EDGE_RESP_CAP)
88 tsv_serve_one(server,input,output,EDGE_REQ_MAX)
89 sys_exit_group(0)
90 }
91 sys_close(server)
92 if pid < 0 { sys_close(client); gv_check("request child fork" as *u8,0,ctr); return 0 }
93 gv_check("complete request submitted" as *u8,ta_journal_write_fd(client,req,rn) == 0,ctr)
94 let response: *u8=sys_mmap(EDGE_RESP_CAP)
95 var response_n: i64=0
96 if disconnected == 1 {
97 // Linux linger{on=1,seconds=0}: close emits RST. Require actual option success.
98 let linger: *u8=sys_mmap(TG_LINGER_BYTES)
99 linger[0]=1 as u8
100 gv_check("RST option accepted" as *u8,sys_setsockopt(client,TG_SOL_SOCKET,TG_SO_LINGER,linger,TG_LINGER_BYTES) == 0,ctr)
101 } else {
102 while response_n < EDGE_RESP_CAP {
103 let r: i64=sys_read(client,((response as i64)+response_n) as *u8,EDGE_RESP_CAP-response_n)
104 if r <= 0 { break }; response_n=response_n+r
105 }
106 }
107 sys_close(client)
108 gv_check("request child actually reaped with zero status" as *u8,tg_wait(pid) == 0,ctr)
109 let capture: *u8=sys_mmap(SYS_PATH_MAX)
110 var cp: i64=ta_cat(capture,0,tool); cp=ta_cat(capture,cp,".response" as *u8); capture[cp]=0 as u8
111 gv_check("response capture retained" as *u8,tg_put(capture,response,response_n,TG_FILE) == 0,ctr)
112 if disconnected == 0 {
113 gv_check("HTTP response received without cap truncation" as *u8,response_n < EDGE_RESP_CAP && tg_has(response,response_n,"HTTP/1.1 200" as *u8) == 1,ctr)
114 if async == 0 { if ta_streq_n(tool,ta_slen(tool),"timing_promote") == 0 {
115 gv_check("actual fixture stdout returned" as *u8,tg_has(response,response_n,"timing-fixture-output" as *u8),ctr)
116 } }
117 }
118 let sizes: *i64=sys_mmap(8) as *i64
119 let start: i64=sys_now_ms()
120 var observed: i64=0
121 while sys_now_ms()-start < tg_timeout {
122 let journal: *u8=sys_read_file("knowledge/status/actlog.jrnl" as *u8,sizes)
123 if (journal as i64) > 0 {
124 observed=tg_rowhas(journal,sizes[0],tool,phase,"timing_v=1" as *u8)
125 sys_free_file(journal,sizes[0])
126 }
127 if observed == 1 { break }; sys_sleep_ms(tg_step)
128 }
129 gv_check("actual selected phase observed in existing journal" as *u8,observed,ctr)
130 let journal: *u8=sys_read_file("knowledge/status/actlog.jrnl" as *u8,sizes)
131 if (journal as i64) > 0 {
132 gv_check("first-byte scope remains unobserved" as *u8,tg_rowhas(journal,sizes[0],tool,phase,"client_first_byte=unobserved result_first_byte=unobserved" as *u8),ctr)
133 if disconnected == 1 {
134 gv_check("failed transport does not report accepted bytes" as *u8,tg_rowhas(journal,sizes[0],tool,"response_write_return" as *u8,"bytes=-1" as *u8),ctr)
135 }
136 if ta_streq_n(tool,ta_slen(tool),"timing_fail") == 1 {
137 gv_check("actual nonzero tool exit is preserved" as *u8,tg_rowhas(journal,sizes[0],tool,"sync_runner_return" as *u8,"code=7" as *u8),ctr)
138 gv_check("positive organ exit retains existing valid-result envelope" as *u8,tg_has(response,response_n,"\"isError\":false" as *u8),ctr)
139 gv_check("positive organ exit is reported exactly in response metadata" as *u8,tg_has(response,response_n,"\"exit_code\":7" as *u8),ctr)
140 gv_check("nonzero-exit fixture keeps its complete output count" as *u8,tg_has(response,response_n,"\"bytes\":22" as *u8),ctr)
141 }
142 var job_case: i64=async
143 if ta_streq_n(tool,ta_slen(tool),"timing_promote") == 1 { job_case=1 }
144 if job_case == 1 {
145 let at: i64=ta_indexof(response,response_n,"\"job\":" as *u8)
146 var job: i64=0; var pos: i64=at+6; var digits: i64=0
147 if at >= 0 {
148 while pos < response_n {
149 let digit: i64=response[pos] as i64
150 if digit < 48 || digit > 57 { break }
151 if job > (TG_I64_MAX-(digit-48))/10 { digits=0; break }
152 job=job*10+digit-48; pos=pos+1; digits=digits+1
153 }
154 }
155 gv_check("actual ACK contains a positive job ID" as *u8,digits > 0 && job > 0,ctr)
156 if digits > 0 && job > 0 {
157 let token: *u8=sys_mmap(TA_MAGIC_4096)
158 var t: i64=ta_cat(token,0," job=" as *u8); t=ta_catn(token,t,job); t=ta_cat(token,t," " as *u8); token[t]=0 as u8
159 gv_check("publication phase correlates to actual ACK job" as *u8,tg_rowhas(journal,sizes[0],tool,phase,token),ctr)
160 var ack: *u8="async_ack_write_return" as *u8
161 if async == 0 { ack="promoted_ack_write_return" as *u8 }
162 gv_check("ACK timing correlates to the same actual job" as *u8,tg_rowhas(journal,sizes[0],tool,ack,token),ctr)
163 let path: *u8=sys_mmap(SYS_PATH_MAX)
164 var p: i64=ta_cat(path,0,"_jobs/job_" as *u8); p=ta_catn(path,p,job)
165 let base: i64=p
166 p=ta_cat(path,p,".out" as *u8); path[p]=0 as u8
167 let jsize: *i64=sys_mmap(8) as *i64
168 let output: *u8=sys_read_file(path,jsize)
169 var exact: i64=0
170 if (output as i64) > 0 { if jsize[0] == 22 { exact=ta_streq_n(output,jsize[0],"timing-fixture-output\n") }; sys_free_file(output,jsize[0]) }
171 gv_check("actual published job output is complete fixture bytes" as *u8,exact,ctr)
172 p=ta_cat(path,base,".claim" as *u8); path[p]=0 as u8
173 let marker: *u8=sys_read_file(path,jsize)
174 gv_check("actual terminal claim exists" as *u8,tg_has(marker,jsize[0],"state=DONE" as *u8),ctr)
175 if async == 0 {
176 gv_check("promotion marker retains unreaped scope" as *u8,tg_has(marker,jsize[0],"PROMOTED-UNREAPED" as *u8),ctr)
177 gv_check("capture observation never invents worker exit" as *u8,tg_rowhas(journal,sizes[0],tool,"promoted_capture_return_unreaped" as *u8,"code=-1" as *u8),ctr)
178 }
179 if (marker as i64) > 0 { sys_free_file(marker,jsize[0]) }
180 }
181 }
182 sys_free_file(journal,sizes[0])
183 }
184 gv_puts("case=" as *u8); gv_puts(tool); gv_puts(" response_bytes=" as *u8); gv_num(response_n); gv_puts("\n" as *u8)
185 return observed
186}
187func main(argc: i64, argv: *i64) -> i64 {
188 if argc != 9 { gv_puts("usage: gate /tmp/fresh fixture-elf port timeout-ms poll-ms window-ms reserve-ms slow-ms\n" as *u8); return 3 }
189 let dir: *u8=argv[1] as *u8
190 let elf: *u8=argv[2] as *u8
191 // The caller supplies the hash-verified, already execution-tested original fixture.
192 // A byte-identical /tmp copy returned EACCES; no mount-policy inference or change.
193 let elf_n: i64=ta_slen(elf)
194 if elf_n <= 1 || elf_n >= SYS_PATH_MAX { return 3 }
195 if elf[0] != (47 as u8) { return 3 }
196 var ep: i64=0
197 while ep < elf_n {
198 if elf[ep] < (32 as u8) || elf[ep] == (127 as u8) { return 3 }
199 ep=ep+1
200 }
201 let port: i64=tg_num(argv[3] as *u8)
202 tg_timeout=tg_num(argv[4] as *u8); tg_step=tg_num(argv[5] as *u8)
203 let window: i64=tg_num(argv[6] as *u8); let reserve: i64=tg_num(argv[7] as *u8); let slow: i64=tg_num(argv[8] as *u8)
204 if ta_slen(dir) >= SYS_PATH_MAX { return 3 }
205 if ta_slen(dir) <= 5 { return 3 }
206 if ta_streq_n(dir,5,"/tmp/") == 0 { return 3 }
207 var i: i64=5
208 while dir[i] != (0 as u8) { if dir[i] == (47 as u8) { return 3 }; if dir[i] < (33 as u8) { return 3 }; i=i+1 }
209 if dir[5] == (46 as u8) { return 3 }
210 if port <= 0 || port > TG_PORT_MAX { return 3 }
211 if reserve < 0 || window <= reserve || slow <= window || tg_timeout <= slow || tg_step <= 0 || tg_step >= tg_timeout { return 3 }
212 if (window-reserve)/2 <= 0 { return 3 }
213 let ctr: *i64=gv_ctr(); gv_head("STAGED SERVE TIMING / actual pinned fixture processes" as *u8)
214 let sz: *i64=sys_mmap(8) as *i64
215 let fixture: *u8=sys_read_file(elf,sz)
216 if (fixture as i64) <= 0 { return 3 }
217 if sz[0] < 4 { return 3 }
218 if fixture[0] != (127 as u8) || fixture[1] != (69 as u8) || fixture[2] != (76 as u8) || fixture[3] != (70 as u8) { return 3 }
219 if sys_mkdir(dir,TG_PRIVATE) != 0 { return 3 }
220 if sys_chdir(dir) != 0 { return 3 }
221 if sys_mkdir("knowledge" as *u8,TG_PRIVATE) != 0 { return 3 }
222 if sys_mkdir("knowledge/status" as *u8,TG_PRIVATE) != 0 { return 3 }
223 if tg_put("fixture.elf" as *u8,fixture,sz[0],TG_EXEC) != 0 { return 3 }
224 if tg_put("knowledge/status/actlog.jrnl" as *u8,"" as *u8,0,TG_FILE) != 0 { return 3 }
225 if tg_put(TA_CAP_KEYFILE,TG_SECRET,ta_slen(TG_SECRET),TG_FILE) != 0 { return 3 }
226 let key_size: *i64=sys_mmap(8) as *i64
227 let key: *u8=ta_load_cap_secret_from(TA_CAP_KEYFILE,key_size)
228 let key_ok: i64=ta_streq_n(key,key_size[0],TG_SECRET)
229 gv_check("actual key loader reads exact gate-only signing key" as *u8,key_ok,ctr)
230 if key_ok != 1 { return gv_verdict("TOOLS-TIMING-SERVE" as *u8,ctr,"fixture key mismatch; no case dispatched" as *u8) }
231 // Five path copies plus named row text and three full-width numeric fields.
232 let conf_cap: i64=5*SYS_PATH_MAX+TA_MAGIC_4096
233 let conf: *u8=sys_mmap(conf_cap)
234 if (conf as i64) <= 0 { return 3 }
235 var n: i64=ta_cat(conf,0,"timing_ok\t" as *u8); n=ta_cat(conf,n,elf)
236 n=ta_cat(conf,n,"\tGREEN\tok 0\ntiming_fail\t" as *u8); n=ta_cat(conf,n,elf)
237 n=ta_cat(conf,n,"\tGREEN\tfail 0\ntiming_async\t" as *u8); n=ta_cat(conf,n,elf)
238 n=ta_cat(conf,n,"\tGREEN\tok " as *u8); n=ta_catn(conf,n,slow)
239 n=ta_cat(conf,n,"\ntiming_promote\t" as *u8); n=ta_cat(conf,n,elf)
240 n=ta_cat(conf,n,"\tGREEN\tok " as *u8); n=ta_catn(conf,n,slow)
241 n=ta_cat(conf,n,"\ntiming_disconnect\t" as *u8); n=ta_cat(conf,n,elf)
242 n=ta_cat(conf,n,"\tGREEN\tok " as *u8); n=ta_catn(conf,n,(window-reserve)/2); n=ta_cat(conf,n,"\n" as *u8)
243 if tg_put(TEA_CONF,conf,n,TG_FILE) != 0 { return 3 }
244 n=ta_cat(conf,0,"edge_window_ms=" as *u8); n=ta_catn(conf,n,window); n=ta_cat(conf,n,"\nreply_reserve_ms=" as *u8); n=ta_catn(conf,n,reserve); n=ta_cat(conf,n,"\n" as *u8)
245 if tg_put(TEA_WINDOW_CONF,conf,n,TG_FILE) != 0 { return 3 }
246 let addr: *u8=sys_mmap(16); nx_http_server_addr_loopback(addr,port)
247 let verdict: *i64=sys_mmap(8) as *i64
248 let signal_rc: i64=sys_ignore_sigpipe()
249 gv_check("live serving SIGPIPE disposition installed" as *u8,signal_rc == 0,ctr)
250 if signal_rc != 0 { return gv_verdict("TOOLS-TIMING-SERVE" as *u8,ctr,"SIGPIPE setup failed; no loopback case dispatched" as *u8) }
251 let lfd: i64=nx_http_server_listen(addr,1,verdict)
252 if lfd < 0 { return 3 }
253 tg_case(lfd,addr,"timing_ok" as *u8,0,0,"sync_runner_return" as *u8,ctr)
254 tg_case(lfd,addr,"timing_fail" as *u8,0,0,"sync_runner_return" as *u8,ctr)
255 tg_case(lfd,addr,"timing_async" as *u8,1,0,"job_publish_return" as *u8,ctr)
256 tg_case(lfd,addr,"timing_promote" as *u8,0,0,"job_publish_return_unreaped" as *u8,ctr)
257 tg_case(lfd,addr,"timing_disconnect" as *u8,0,1,"response_write_return" as *u8,ctr)
258 sys_close(lfd)
259 gv_puts("retained_fixture=" as *u8); gv_puts(dir); gv_puts("\n" as *u8)
260 return gv_verdict("TOOLS-TIMING-SERVE" as *u8,ctr,"actual source-linked serve; no main-daemon seed; promoted worker exit remains unreaped" as *u8)
261}