nx_mcp_stdio_candidate_t230.nx source
↩ module page · 418 lines · 22229 B
1// MCP stdio to sovereign HTTPS. The server remains the authority for discovery and capability checks.
2// Each request runs in a child so the current TLS allocator cannot accumulate across a long session.
3//
4// FAILURE SEMANTICS (lane conn, 2026-09-11). A per-request failure -- a child that exits non-zero, the
5// SIGALRM deadline, a non-2xx HTTP status, an over-budget response, a response-id mismatch, or a
6// TLS/transport failure -- produces a JSON-RPC 2.0 error response for THAT request id on the protocol
7// fd, and the connector KEEPS SERVING every other request. Two mechanisms, one shape:
8// * the request CHILD self-reports soft failures (it knows the http status / byte limit) by writing a
9// correlated error to its own pipe and exiting 0, so the parent forwards it like any response;
10// * the PARENT (mp_recover_pump) turns a child that died WITHOUT self-reporting -- a crash, or the
11// SIGALRM deadline killing it -- into a correlated error from that slot's pending id and continues.
12// An oversized OR malformed request line is consumed to its newline and answered with -32600 for its id
13// when the id is readable from the retained prefix; otherwise a stderr diagnostic. Notifications (no id)
14// never get a response. Only broken stdin/stdout, poll failure, or allocation failure ends the process.
15// Protocol-fd isolation (stdout dup'd away to stderr for library chatter, the real protocol pipe held on
16// a private fd) and response-id correlation are preserved. No magic numbers: every code/limit is named.
17import "nx_mcp_transport.nx"
18import "nx_mcp_route.nx"
19import "nx_request_pool_candidate_t230.nx"
20import "nx_mcp_control.nx"
21import "nx_mcp_pending_candidate_t230.nx"
22
23const MS_F_DUPFD_CLOEXEC: i64 = 1030
24const MS_FIRST_PRIVATE_FD: i64 = 3
25const MS_STDIN: i64 = 0
26const MS_STDOUT: i64 = 1
27const MS_STDERR: i64 = 2
28
29// ms_request return codes. MS_DUP is < 0 so a bare `rc != 0` still catches it, but the caller checks it
30// FIRST and answers the one duplicate request instead of ending the session; MS_FATAL is a resource /
31// allocation failure that legitimately ends the process.
32const MS_OK: i64 = 0
33const MS_DUP: i64 = 0 - 1
34const MS_FATAL: i64 = 10
35
36// Per-stage JSON-RPC error codes. Soft failures the CHILD detects use the server range -32000..-32099;
37// the request-line codes are the standard -32600 (Invalid Request). Named, never bare.
38const MSE_CODE_TRANSPORT: i64 = 0 - 32001
39const MSE_CODE_HTTPSTATUS: i64 = 0 - 32002
40const MSE_CODE_PARSE: i64 = 0 - 32003
41const MSE_CODE_BODY: i64 = 0 - 32004
42const MSE_CODE_RESPSIZE: i64 = 0 - 32005
43const MSE_CODE_IDENTITY: i64 = 0 - 32006
44const MSE_CODE_TRUSTSTORE: i64 = 0 - 32007
45const MSE_CODE_CREDENTIAL: i64 = 0 - 32008
46const MSE_CODE_URL: i64 = 0 - 32009
47const MSE_CODE_ROUTE: i64 = 0 - 32010
48const MSE_CODE_PROTOWRITE: i64 = 0 - 32011
49const MSO_CODE_DUP: i64 = 0 - 32013
50const MSO_CODE_INVALID: i64 = 0 - 32600
51
52// A borrowed failure descriptor the child fills from ms_exchange, then hands to the one correlated-error
53// emitter. Stage/cause/next are static literals (no user bytes, so no escaping); measured is the number
54// the reader needs -- the http status, the byte budget, a transport code.
55struct NxMcpErr { code: i64, stage: *u8, cause: *u8, measured: i64, next: *u8, http_data:*u8, http_n:i64 }
56
57func ms_number(s: *u8) -> i64 {
58 var n: i64 = 0
59 var i: i64 = 0
60 while s[i] != (0 as u8) {
61 let c: i64 = s[i] as i64
62 if c < 48 { return 0 }
63 if c > 57 { return 0 }
64 if n > (NX_RA_SIZE_MAX - (c - 48)) / 10 { return 0 }
65 n = n * 10 + c - 48
66 i = i + 1
67 }
68 return n
69}
70
71// Locate a top-level id without confusing a nested tools argument or escaped string with the envelope.
72func ms_has_id(src: *u8, n: i64) -> i64 {
73 var depth: i64 = 0
74 var quoted: i64 = 0
75 var escaped: i64 = 0
76 var start: i64 = 0
77 var i: i64 = 0
78 while i < n {
79 let c: i64 = src[i] as i64
80 if quoted == 1 {
81 if escaped == 1 { escaped = 0 }
82 else { if c == 92 { escaped = 1 }
83 else { if c == 34 {
84 quoted = 0
85 if depth == 1 { if i - start == 2 {
86 if src[start] == (105 as u8) { if src[start + 1] == (100 as u8) {
87 var j: i64 = i + 1
88 while j < n { if src[j] == (32 as u8) { j = j + 1 } else { break } }
89 if j < n { if src[j] == (58 as u8) { return 1 } }
90 } }
91 } }
92 } } }
93 } else {
94 if c == 34 { quoted = 1; start = i + 1 }
95 else { if c == 123 { depth = depth + 1 }
96 else { if c == 125 { depth = depth - 1 } } }
97 }
98 i = i + 1
99 }
100 return 0
101}
102
103// Perform one request in a child, writing the response (or NOTHING for a notification) to protocol_fd.
104// On failure it fills `err` and returns a non-zero code WITHOUT writing to protocol_fd, so the caller
105// owns the one correlated-error write. The protocol_fd is the child's pipe to the parent.
106func ms_exchange(base: *u8, capfile: *u8, body: *u8, body_n: i64, response_cap: i64, protocol_fd: i64, timeout_secs: i64, err: *NxMcpErr) -> i64 {
107 let store_rc: i64 = nx_trust_store_load_from_certdata("data/mozilla_certdata.txt" as *u8, 512, 4194304)
108 if store_rc <= 0 {
109 err.code = MSE_CODE_TRUSTSTORE; err.stage = "trust-store" as *u8
110 err.cause = "the TLS trust store failed to load" as *u8; err.measured = store_rc
111 err.next = "verify data/mozilla_certdata.txt is present and readable" as *u8
112 return 1
113 }
114 let cap: *u8 = sys_mmap(8192)
115 let cap_n_raw: i64 = mc_read_file(capfile, cap, 8191)
116 if cap_n_raw <= 0 {
117 err.code = MSE_CODE_CREDENTIAL; err.stage = "credential" as *u8
118 err.cause = "the routed capability file is unreadable or empty" as *u8; err.measured = cap_n_raw
119 err.next = "check the routed cap file path and permissions" as *u8
120 return 2
121 }
122 let cap_n: i64 = mc_rtrim_nl(cap, cap_n_raw)
123 let full: *u8 = sys_mmap(4096)
124 mc_join_url(base, "/mcp" as *u8, full)
125 let target: *NxHttpsTarget = sys_mmap(64) as *NxHttpsTarget
126 target.url = nx_url_new(); target.port = 0
127 if nx_https_url_for_fetch(full, target) != NX_HTTPS_URL_OK {
128 err.code = MSE_CODE_URL; err.stage = "url" as *u8
129 err.cause = "the base URL could not be parsed for fetch" as *u8; err.measured = 0
130 err.next = "verify the base URL argument" as *u8
131 return 3
132 }
133 let req: *u8 = sys_mmap(body_n + 16384)
134 let req_n: i64 = mc_build_request("POST" as *u8, 4,
135 full + target.url.path_off, target.url.path_len,
136 full + target.url.host_off, target.url.host_len,
137 cap, cap_n, "application/json" as *u8, 16, body, body_n, req)
138 let out: *u8 = sys_mmap(response_cap)
139 let n: i64 = mc_req_timed(store_rc as *TrustStore, full, target, req, req_n, out, response_cap, timeout_secs)
140 if n <= 0 {
141 err.code = MSE_CODE_TRANSPORT; err.stage = "transport" as *u8
142 err.cause = "the TLS or HTTP request failed or the connection closed" as *u8; err.measured = n
143 err.next = "check network reachability, TLS trust and service health" as *u8
144 return 4
145 }
146 if n >= response_cap {
147 err.code = MSE_CODE_RESPSIZE; err.stage = "response-size" as *u8
148 err.cause = "the response exceeded the configured response byte budget" as *u8; err.measured = response_cap
149 err.next = "raise the response byte budget or narrow the request" as *u8
150 return 5
151 }
152 let parsed: *i64 = nx_http_resp_alloc()
153 if nx_http_response_parse(out, n, parsed) != 0 {
154 err.code = MSE_CODE_PARSE; err.stage = "response-parse" as *u8
155 err.cause = "the HTTP response could not be parsed" as *u8; err.measured = n
156 err.next = "inspect the upstream response framing" as *u8
157 return 6
158 }
159 if parsed[1] < 200||parsed[1] >= 300 {
160 err.code = MSE_CODE_HTTPSTATUS; err.stage = "http-status" as *u8
161 err.cause = "the upstream returned a non-2xx HTTP status" as *u8; err.measured = parsed[1]
162 err.next = "inspect selective HTTP evidence and service health; verify outcome before any write retry" as *u8
163 err.http_data=mpe_http_evidence(out,n,parsed,&err.http_n)
164 return 7
165 }
166 // Notifications have no response on stdio, even if the remote endpoint returns an acknowledgement body.
167 if ms_has_id(body, body_n) == 0 { return 0 }
168 let off: i64 = parsed[6]
169 let count: i64 = n - off
170 if parsed[8] == NX_HTTP_BODY_CONTENT_LENGTH { if count != parsed[7] {
171 err.code = MSE_CODE_BODY; err.stage = "response-body" as *u8
172 err.cause = "the response body was truncated against its declared length" as *u8; err.measured = count
173 err.next = "verify the upstream response is a complete JSON object" as *u8
174 return 8
175 } }
176 if count < 2 {
177 err.code = MSE_CODE_BODY; err.stage = "response-body" as *u8
178 err.cause = "the response body is too short to be a JSON object" as *u8; err.measured = count
179 err.next = "verify the upstream response is a complete JSON object" as *u8
180 return 8
181 }
182 if out[off] != (123 as u8) {
183 err.code = MSE_CODE_BODY; err.stage = "response-body" as *u8
184 err.cause = "the response body is not a JSON object" as *u8; err.measured = count
185 err.next = "verify the upstream response is a complete JSON object" as *u8
186 return 8
187 }
188 if mr_response_matches(body, body_n, out + off, count) != 1 {
189 err.code = MSE_CODE_IDENTITY; err.stage = "response-identity" as *u8
190 err.cause = "the response id is absent, ambiguous, invalid or different from the request id" as *u8; err.measured = 0
191 err.next = "inspect server request/response correlation; verify write outcome before retrying" as *u8
192 return 14
193 }
194 if mc_write_n(protocol_fd, out + off, count) != 0 {
195 err.code = MSE_CODE_PROTOWRITE; err.stage = "protocol-write" as *u8
196 err.cause = "writing the response to the protocol channel failed" as *u8; err.measured = 0
197 err.next = "the downstream reader closed; reconnect the session" as *u8
198 return 9
199 }
200 if mc_write_n(protocol_fd, "\n" as *u8, 1) != 0 {
201 err.code = MSE_CODE_PROTOWRITE; err.stage = "protocol-write" as *u8
202 err.cause = "writing the response terminator to the protocol channel failed" as *u8; err.measured = 0
203 err.next = "the downstream reader closed; reconnect the session" as *u8
204 return 9
205 }
206 return 0
207}
208
209// Fork a request worker. In the PARENT return MS_OK (forked), MS_DUP (id already in flight -- the caller
210// answers that one request and continues), or MS_FATAL (fork/allocation failure -- ends the process).
211// In the CHILD run the request and, on ANY failure, write ONE correlated error to the pipe then exit 0
212// so the parent forwards it and the session survives; a notification (no id) self-reports NOTHING.
213func ms_request(base: *u8, capfile: *u8, body: *u8, body_n: i64, response_cap: i64, protocol_fd: i64, timeout_secs: i64, argc: i64, argv: *i64, route_first: i64, pool: *NxRequestPool, pending: *u8, control: *NxMcpControl) -> i64 {
214 let pid: i64 = mp_spawn(pool, pending, body, control)
215 if pid == 0 {
216 // CHILD. The deadline is a backstop: on expiry SIGALRM kills this child, the parent observes the
217 // dead worker and emits the correlated deadline error (mp_recover_pump), so the session survives.
218 sys_alarm(timeout_secs)
219 let err: *NxMcpErr = sys_mmap(__size_of(NxMcpErr)) as *NxMcpErr
220 var selected: *u8 = capfile
221 if argc > route_first {
222 let slot: i64 = mr_select(body, body_n, argc, argv, route_first, 2)
223 if slot < 0 {
224 mpe_write_error(pool.child_fd, control.id_kind, ((body as i64) + control.id_off) as *u8, control.id_len, 0,
225 MSE_CODE_ROUTE, "credential-selection" as *u8,
226 "malformed, ambiguous or unsupported routing selector" as *u8, 0,
227 "use unique unescaped method, params and name keys; inspect connector routing qualification" as *u8)
228 sys_exit(0)
229 return 0
230 }
231 selected = argv[slot] as *u8
232 }
233 let rc: i64 = ms_exchange(base, selected, body, body_n, response_cap, pool.child_fd, timeout_secs, err)
234 if rc != 0 {
235 mpe_write_error_detail(pool.child_fd, control.id_kind, ((body as i64) + control.id_off) as *u8, control.id_len, 0,
236 err.code, err.stage, err.cause, err.measured, err.next, err.http_data, err.http_n)
237 sys_exit(0)
238 return 0
239 }
240 sys_exit(0)
241 return 0
242 }
243 if pid == MP_SPAWN_DUP { return MS_DUP }
244 if pid < 0 { return -2 }
245 return pid
246}
247
248func ms_drain(pool: *NxRequestPool, pending: *u8, protocol_fd: i64, timeout_secs: i64) -> i64 {
249 while pool.active > 0 {
250 if mp_recover_pump(pool, pending, 0, protocol_fd, timeout_secs) < 0 { return -1 }
251 }
252 return 0
253}
254
255
256func ms_dispatch_queue(q:*NxMcpQueue,pool:*NxRequestPool,pending:*u8,protocol_fd:i64,timeout_secs:i64,response_cap:i64,argc:i64,argv:*i64,route_first:i64)->i64 {
257 if mq_expire(q,sys_now_ms(),protocol_fd)!=0{return -1}
258 while mq_ready(q,pool)!=0{
259 let node:*NxMcpQueued=q.head
260 let pid:i64=ms_request(argv[1] as *u8,argv[2] as *u8,node.body,node.body_n,response_cap,protocol_fd,timeout_secs,argc,argv,route_first,pool,pending,node.control)
261 mq_unlink(q,0 as *NxMcpQueued,node)
262 if pid<0{
263 if mq_error(node,protocol_fd,MQ_CODE_CAPACITY,"worker could not be created; not dispatched",pid)!=0{mq_free(node);return -1}
264 mq_event(q,node,"refused-worker-allocation",0)
265 }else{
266 if node.barrier!=0{q.barrier_pid=pid}
267 mq_event(q,node,"dispatched",1)
268 }
269 mq_free(node)
270 }
271 return 0
272}
273
274func main(argc: i64, argv: *i64) -> i64 {
275 var route_first:i64=7
276 var workers:i64=1
277 var queue_bytes:i64=0
278 var queue_wait_ms:i64=0
279 var options:i64=0
280 while route_first<argc{
281 let name:*u8=argv[route_first] as *u8
282 var flag:i64=0
283 if mr_equal(name,mr_len(name),"--workers")==1{flag=1}
284 if mr_equal(name,mr_len(name),"--queue-bytes")==1{flag=2}
285 if mr_equal(name,mr_len(name),"--queue-timeout-ms")==1{flag=4}
286 if flag==0{break}
287 if (options&flag)!=0||route_first+1>=argc{return 1}
288 let value:i64=ms_number(argv[route_first+1] as *u8)
289 if value<=0{return 1}
290 if flag==1{workers=value}
291 if flag==2{queue_bytes=value}
292 if flag==4{queue_wait_ms=value}
293 options=options|flag;route_first=route_first+2
294 }
295 if mr_config(argc, argv, route_first) != 1 {
296 mc_puts("usage: nx_mcp_stdio <base_url> <cap_file> <request_bytes> <response_bytes> <read_chunk_bytes> <request_timeout_seconds> [--workers <admitted_worker_count>] [--queue-bytes <memory_budget>] [--queue-timeout-ms <wait_budget>] [<tool_name> <cap_file> ...]; tool routes must be unique\n" as *u8)
297 return 1
298 }
299 let request_cap: i64 = ms_number(argv[3] as *u8)
300 let response_cap: i64 = ms_number(argv[4] as *u8)
301 let chunk_cap: i64 = ms_number(argv[5] as *u8)
302 let timeout_secs: i64 = ms_number(argv[6] as *u8)
303 if timeout_secs < 1 { return 1 }
304 if request_cap < 1 { return 1 }
305 if response_cap < 1 { return 1 }
306 if chunk_cap < 1 { return 1 }
307 if chunk_cap > request_cap { return 1 }
308 if workers > NX_RA_SIZE_MAX / __size_of(NxMcpPending) { return 1 }
309 let default_queue_bytes:i64=mq_default_budget(workers,request_cap)
310 if default_queue_bytes<=0{return 1}
311 if timeout_secs>NX_RA_SIZE_MAX/1000{return 1}
312 // Bootstrap queue envelope derives from configured request memory and time,
313 // independently overridable; executor capacity remains unchanged.
314 if queue_bytes==0{queue_bytes=default_queue_bytes}
315 if queue_wait_ms==0{queue_wait_ms=timeout_secs*1000}
316 let queue:*NxMcpQueue=mq_new(queue_bytes,queue_wait_ms)
317 if queue==(0 as *NxMcpQueue){return 1}
318 // Reserve the protocol pipe before redirecting every library diagnostic to stderr.
319 let protocol_fd: i64 = __syscall(NX_SYS_FCNTL, MS_STDOUT, MS_F_DUPFD_CLOEXEC, MS_FIRST_PRIVATE_FD, 0, 0, 0)
320 if protocol_fd < MS_FIRST_PRIVATE_FD { return 1 }
321 if sys_dup3(MS_STDERR, MS_STDOUT, 0) < 0 { return 1 }
322 let body: *u8 = sys_mmap(request_cap)
323 let chunk: *u8 = sys_mmap(chunk_cap)
324 let control: *NxMcpControl = sys_mmap(__size_of(NxMcpControl)) as *NxMcpControl
325 let pool: *NxRequestPool = rp_new(workers,response_cap)
326 if pool == (0 as *NxRequestPool) { return 1 }
327 let pending: *u8 = sys_mmap(workers*__size_of(NxMcpPending))
328 var input_closed:i64=0
329 var used: i64 = 0
330 var overflow: i64 = 0
331 while true {
332 if ms_dispatch_queue(queue,pool,pending,protocol_fd,timeout_secs,response_cap,argc,argv,route_first)!=0{rp_cancel_all(pool);return 12}
333 if input_closed!=0&&pool.active==0&&queue.count==0{return 0}
334 if pool.active>0||queue.count>0{
335 let ready:i64=mp_recover_pump_timeout(pool,pending,1-input_closed,protocol_fd,timeout_secs,mq_wait(queue,sys_now_ms()))
336 if ready<0{rp_cancel_all(pool);return 12}
337 if ready==0{continue}
338 }
339 if input_closed!=0{continue}
340 let n:i64=sys_read(MS_STDIN,chunk,chunk_cap)
341 if n==RP_EINTR{continue}
342 if n<0{rp_cancel_all(pool);return 2}
343 if n==0{
344 if used!=0||overflow!=0{rp_cancel_all(pool);return 3}
345 input_closed=1;continue
346 }
347 var i: i64 = 0
348 while i < n {
349 let c: u8 = chunk[i]
350 if overflow != 0 {
351 // Consuming an oversized request line to its newline. On the newline, answer the id we
352 // retained (if any) with -32600, else a stderr diagnostic, and keep serving.
353 if c == (10 as u8) {
354 var okind: i64 = 0
355 var ooff: i64 = 0
356 var olen: i64 = 0
357 if mo_prefix_id(body, used, &okind, &ooff, &olen) == 1 {
358 if mpe_write_error(protocol_fd, okind, ((body as i64) + ooff) as *u8, olen, 0,
359 MSO_CODE_INVALID, "request-line" as *u8,
360 "the request line exceeded the configured byte budget" as *u8, request_cap,
361 "split the request or raise the request byte budget" as *u8) != 0 { rp_cancel_all(pool); return 12 }
362 } else {
363 mc_puts("{\"owner\":\"nx_mcp_stdio\",\"stage\":\"request-line\",\"cause\":\"request line exceeded the byte budget and no id was readable from the retained prefix\",\"impact\":\"line discarded, no response correlated\",\"next\":\"send a smaller request or raise the request byte budget\"}\n" as *u8)
364 }
365 used = 0
366 overflow = 0
367 }
368 i = i + 1
369 continue
370 }
371 if c == (10 as u8) {
372 if used > 0 {
373 if mct_read(body,used,control) != 0 {
374 // A malformed request line does not end the session: answer -32600 for the id if
375 // one is readable from the raw line, else a stderr diagnostic, then keep serving.
376 var mkind: i64 = 0
377 var moff: i64 = 0
378 var mlen: i64 = 0
379 if mo_prefix_id(body, used, &mkind, &moff, &mlen) == 1 {
380 if mpe_write_error(protocol_fd, mkind, ((body as i64) + moff) as *u8, mlen, 0,
381 MSO_CODE_INVALID, "control-envelope" as *u8,
382 "malformed or ambiguous request metadata" as *u8, used,
383 "correct the JSON-RPC envelope and resend" as *u8) != 0 { rp_cancel_all(pool); return 12 }
384 } else {
385 mc_puts("{\"owner\":\"nx_mcp_stdio\",\"stage\":\"control-envelope\",\"cause\":\"malformed request metadata and no id was readable\",\"impact\":\"line discarded, no response correlated\",\"next\":\"correct the JSON-RPC envelope and resend\"}\n" as *u8)
386 }
387 used = 0
388 i = i + 1
389 continue
390 }
391 if control.method == MCT_CANCEL {
392 var cancelled:i64=mq_cancel(queue,body,control,protocol_fd)
393 if cancelled==1{used=0;i=i+1;continue}
394 if cancelled== -2{rp_cancel_all(pool);return 12}
395 if cancelled==0{cancelled=mp_cancel(pool,pending,body,control)}
396 if cancelled < 0 {
397 // A cancel that cannot be honoured (an initialize, or a reap fault) is not a
398 // reason to end the session. notifications/cancelled has no id, so no response.
399 mc_puts("{\"owner\":\"nx_mcp_stdio\",\"stage\":\"cancellation\",\"cause\":\"cancellation could not be applied (unknown, protected or reap fault)\",\"impact\":\"no worker was cancelled\",\"next\":\"reconcile the target request before retrying\"}\n" as *u8)
400 }
401 if cancelled > 0 {
402 mc_puts("{\"owner\":\"nx_mcp_stdio\",\"stage\":\"cancellation\",\"local_worker\":\"reaped\",\"remote_job_cancelled\":false,\"next\":\"Reconcile remote operation receipt before retrying a write\"}\n" as *u8)
403 }
404 used=0; i=i+1; continue
405 }
406 let admitted:i64=mq_enqueue(queue,pool,pending,body,used,control,sys_now_ms(),protocol_fd)
407 if admitted<0{rp_cancel_all(pool);return 12}
408 used = 0
409 }
410 } else {
411 if used >= request_cap { overflow = 1; i = i + 1; continue }
412 body[used] = c; used = used + 1
413 }
414 i = i + 1
415 }
416 }
417 return 0
418}