`nx_mcp_stdio.nx` source

↩ module page · 413 lines · 21774 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_mcp_sse.nx"
20import "nx_request_pool.nx"
21import "nx_mcp_control.nx"
22import "nx_mcp_pending.nx"
23
24const MS_F_DUPFD_CLOEXEC: i64 = 1030
25const MS_FIRST_PRIVATE_FD: i64 = 3
26const MS_STDIN: i64 = 0
27const MS_STDOUT: i64 = 1
28const MS_STDERR: i64 = 2
29
30// ms_request return codes. MS_DUP is < 0 so a bare `rc != 0` still catches it, but the caller checks it
31// FIRST and answers the one duplicate request instead of ending the session; MS_FATAL is a resource /
32// allocation failure that legitimately ends the process.
33const MS_OK:    i64 = 0
34const MS_DUP:   i64 = 0 - 1
35const MS_FATAL: i64 = 10
36
37// Per-stage JSON-RPC error codes. Soft failures the CHILD detects use the server range -32000..-32099;
38// the request-line codes are the standard -32600 (Invalid Request). Named, never bare.
39const MSE_CODE_TRANSPORT:  i64 = 0 - 32001
40const MSE_CODE_HTTPSTATUS: i64 = 0 - 32002
41const MSE_CODE_PARSE:      i64 = 0 - 32003
42const MSE_CODE_BODY:       i64 = 0 - 32004
43const MSE_CODE_RESPSIZE:   i64 = 0 - 32005
44const MSE_CODE_IDENTITY:   i64 = 0 - 32006
45const MSE_CODE_TRUSTSTORE: i64 = 0 - 32007
46const MSE_CODE_CREDENTIAL: i64 = 0 - 32008
47const MSE_CODE_URL:        i64 = 0 - 32009
48const MSE_CODE_ROUTE:      i64 = 0 - 32010
49const MSE_CODE_PROTOWRITE: i64 = 0 - 32011
50const MSO_CODE_DUP:        i64 = 0 - 32013
51const MSO_CODE_INVALID:    i64 = 0 - 32600
52
53// A borrowed failure descriptor the child fills from ms_exchange, then hands to the one correlated-error
54// emitter. Stage/cause/next are static literals (no user bytes, so no escaping); measured is the number
55// the reader needs -- the http status, the byte budget, a transport code.
56struct NxMcpErr { code: i64, stage: *u8, cause: *u8, measured: i64, next: *u8, http_data:*u8, http_n:i64 }
57
58func ms_number(s: *u8) -> i64 {
  var n: i64 = 0
  var i: i64 = 0
  while s[i] != (0 as u8) {
      let c: i64 = s[i] as i64
      if c < 48 { return 0 }
      if c > 57 { return 0 }
      if n > (NX_RA_SIZE_MAX - (c - 48)) / 10 { return 0 }
      n = n * 10 + c - 48
      i = i + 1
  }
  return n
70}
71
72// Locate a top-level id without confusing a nested tools argument or escaped string with the envelope.
73func ms_has_id(src: *u8, n: i64) -> i64 {
  var depth: i64 = 0
  var quoted: i64 = 0
  var escaped: i64 = 0
  var start: i64 = 0
  var i: i64 = 0
  while i < n {
      let c: i64 = src[i] as i64
      if quoted == 1 {
          if escaped == 1 { escaped = 0 }
          else { if c == 92 { escaped = 1 }
          else { if c == 34 {
              quoted = 0
              if depth == 1 { if i - start == 2 {
                  if src[start] == (105 as u8) { if src[start + 1] == (100 as u8) {
                      var j: i64 = i + 1
                      while j < n { if src[j] == (32 as u8) { j = j + 1 } else { break } }
                      if j < n { if src[j] == (58 as u8) { return 1 } }
                  } }
              } }
          } } }
      } else {
          if c == 34 { quoted = 1; start = i + 1 }
          else { if c == 123 { depth = depth + 1 }
          else { if c == 125 { depth = depth - 1 } } }
      }
      i = i + 1
  }
  return 0
102}
103
104// Perform one request in a child, writing the response (or NOTHING for a notification) to protocol_fd.
105// On failure it fills `err` and returns a non-zero code WITHOUT writing to protocol_fd, so the caller
106// owns the one correlated-error write. The protocol_fd is the child's pipe to the parent.
107func ms_exchange(base: *u8, capfile: *u8, body: *u8, body_n: i64, response_cap: i64, protocol_fd: i64, timeout_secs: i64, err: *NxMcpErr) -> i64 {
  let store_rc: i64 = nx_trust_store_load_from_certdata("data/mozilla_certdata.txt" as *u8, 512, 4194304)
  if store_rc <= 0 {
      err.code = MSE_CODE_TRUSTSTORE; err.stage = "trust-store" as *u8
      err.cause = "the TLS trust store failed to load" as *u8; err.measured = store_rc
      err.next = "verify data/mozilla_certdata.txt is present and readable" as *u8
      return 1
  }
  let cap: *u8 = sys_mmap(8192)
  let cap_n_raw: i64 = mc_read_file(capfile, cap, 8191)
  if cap_n_raw <= 0 {
      err.code = MSE_CODE_CREDENTIAL; err.stage = "credential" as *u8
      err.cause = "the routed capability file is unreadable or empty" as *u8; err.measured = cap_n_raw
      err.next = "check the routed cap file path and permissions" as *u8
      return 2
  }
  let cap_n: i64 = mc_rtrim_nl(cap, cap_n_raw)
  let full: *u8 = sys_mmap(4096)
  mc_join_url(base, "/mcp" as *u8, full)
  let target: *NxHttpsTarget = sys_mmap(64) as *NxHttpsTarget
  target.url = nx_url_new(); target.port = 0
  if nx_https_url_for_fetch(full, target) != NX_HTTPS_URL_OK {
      err.code = MSE_CODE_URL; err.stage = "url" as *u8
      err.cause = "the base URL could not be parsed for fetch" as *u8; err.measured = 0
      err.next = "verify the base URL argument" as *u8
      return 3
  }
  let req: *u8 = sys_mmap(body_n + 16384)
  let req_n: i64 = mc_build_request("POST" as *u8, 4,
      full + target.url.path_off, target.url.path_len,
      full + target.url.host_off, target.url.host_len,
      cap, cap_n, "application/json" as *u8, 16, body, body_n, req)
  let out: *u8 = sys_mmap(response_cap)
  let n: i64 = mc_req_timed(store_rc as *TrustStore, full, target, req, req_n, out, response_cap, timeout_secs)
  if n <= 0 {
      err.code = MSE_CODE_TRANSPORT; err.stage = "transport" as *u8
      err.cause = "the TLS or HTTP request failed or the connection closed" as *u8; err.measured = n
      err.next = "check network reachability, TLS trust and service health" as *u8
      return 4
  }
  if n >= response_cap {
      err.code = MSE_CODE_RESPSIZE; err.stage = "response-size" as *u8
      err.cause = "the response exceeded the configured response byte budget" as *u8; err.measured = response_cap
      err.next = "raise the response byte budget or narrow the request" as *u8
      return 5
  }
  let parsed: *i64 = nx_http_resp_alloc()
  if nx_http_response_parse(out, n, parsed) != 0 {
      err.code = MSE_CODE_PARSE; err.stage = "response-parse" as *u8
      err.cause = "the HTTP response could not be parsed" as *u8; err.measured = n
      err.next = "inspect the upstream response framing" as *u8
      return 6
  }
  if parsed[1] < 200||parsed[1] >= 300 {
      err.code = MSE_CODE_HTTPSTATUS; err.stage = "http-status" as *u8
      err.cause = "the upstream returned a non-2xx HTTP status" as *u8; err.measured = parsed[1]
      err.next = "inspect selective HTTP evidence and service health; verify outcome before any write retry" as *u8
      err.http_data=mpe_http_evidence(out,n,parsed,&err.http_n)
      return 7
  }
  // Notifications have no response on stdio, even if the remote endpoint returns an acknowledgement body.
  if ms_has_id(body, body_n) == 0 { return 0 }
  let off: i64 = parsed[6]
  let count: i64 = n - off
  if parsed[8] == NX_HTTP_BODY_CONTENT_LENGTH { if count != parsed[7] {
      err.code = MSE_CODE_BODY; err.stage = "response-body" as *u8
      err.cause = "the response body was truncated against its declared length" as *u8; err.measured = count
      err.next = "verify the upstream response is a complete JSON object" as *u8
      return 8
  } }
  let media:i64=mss_media_type(out,n,parsed)
  var payload:*u8=out+off
  var payload_n:i64=count
  if parsed[8]==NX_HTTP_BODY_CHUNKED{
      payload=sys_mmap(count+1)
      payload_n=nx_http_dechunk(out+off,count,payload,count)
  }
  let decoded:*u8=sys_mmap(response_cap)
  var decoded_n:i64=MSS_BAD_BODY
  if media>=0&&payload_n>0{decoded_n=mss_decode(body,body_n,payload,payload_n,media,decoded,response_cap)}
  if decoded_n<=0{
      err.code=MSE_CODE_BODY;err.stage="response-body" as *u8
      if decoded_n==MSS_BAD_ID{err.code=MSE_CODE_IDENTITY;err.stage="response-identity" as *u8}
      err.cause="the JSON or SSE response is incomplete, malformed, unsupported or ambiguously correlated" as *u8
      err.measured=decoded_n
      err.next="inspect response framing and correlation; verify write outcome before retrying" as *u8
      return 8
  }
  if mc_write_n(protocol_fd,decoded,decoded_n)!=0{
      err.code=MSE_CODE_PROTOWRITE;err.stage="protocol-write" as *u8
      err.cause="writing the validated response batch to the protocol channel failed" as *u8;err.measured=0
      err.next="the downstream reader closed; reconnect the session" as *u8
      return 9
  }
  return 0
202}
203
204// Fork a request worker. In the PARENT return MS_OK (forked), MS_DUP (id already in flight -- the caller
205// answers that one request and continues), or MS_FATAL (fork/allocation failure -- ends the process).
206// In the CHILD run the request and, on ANY failure, write ONE correlated error to the pipe then exit 0
207// so the parent forwards it and the session survives; a notification (no id) self-reports NOTHING.
208func 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 {
  let pid: i64 = mp_spawn(pool, pending, body, control)
  if pid == 0 {
      // CHILD. The deadline is a backstop: on expiry SIGALRM kills this child, the parent observes the
      // dead worker and emits the correlated deadline error (mp_recover_pump), so the session survives.
      sys_alarm(timeout_secs)
      let err: *NxMcpErr = sys_mmap(__size_of(NxMcpErr)) as *NxMcpErr
      var selected: *u8 = capfile
      if argc > route_first {
          let slot: i64 = mr_select(body, body_n, argc, argv, route_first, 2)
          if slot < 0 {
              mpe_write_error(pool.child_fd, control.id_kind, ((body as i64) + control.id_off) as *u8, control.id_len, 0,
                  MSE_CODE_ROUTE, "credential-selection" as *u8,
                  "malformed, ambiguous or unsupported routing selector" as *u8, 0,
                  "use unique unescaped method, params and name keys; inspect connector routing qualification" as *u8)
              sys_exit(0)
              return 0
          }
          selected = argv[slot] as *u8
      }
      let rc: i64 = ms_exchange(base, selected, body, body_n, response_cap, pool.child_fd, timeout_secs, err)
      if rc != 0 {
          mpe_write_error_detail(pool.child_fd, control.id_kind, ((body as i64) + control.id_off) as *u8, control.id_len, 0,
              err.code, err.stage, err.cause, err.measured, err.next, err.http_data, err.http_n)
          sys_exit(0)
          return 0
      }
      sys_exit(0)
      return 0
  }
  if pid == MP_SPAWN_DUP { return MS_DUP }
  if pid < 0 { return -2 }
  return pid
241}
242
243func ms_drain(pool: *NxRequestPool, pending: *u8, protocol_fd: i64, timeout_secs: i64) -> i64 {
  while pool.active > 0 {
      if mp_recover_pump(pool, pending, 0, protocol_fd, timeout_secs) < 0 { return -1 }
  }
  return 0
248}
249
250
251func 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 {
if mq_expire(q,sys_now_ms(),protocol_fd)!=0{return -1}
while mq_ready(q,pool)!=0{
let node:*NxMcpQueued=q.head
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)
mq_unlink(q,0 as *NxMcpQueued,node)
if pid<0{
 if mq_error(node,protocol_fd,MQ_CODE_CAPACITY,"worker could not be created; not dispatched",pid)!=0{mq_free(node);return -1}
 mq_event(q,node,"refused-worker-allocation",0)
}else{
 if node.barrier!=0{q.barrier_pid=pid}
 mq_event(q,node,"dispatched",1)
}
mq_free(node)
}
return 0
267}
268
269func main(argc: i64, argv: *i64) -> i64 {
  var route_first:i64=7
  var workers:i64=1
  var queue_bytes:i64=0
  var queue_wait_ms:i64=0
  var options:i64=0
  while route_first<argc{
   let name:*u8=argv[route_first] as *u8
   var flag:i64=0
   if mr_equal(name,mr_len(name),"--workers")==1{flag=1}
   if mr_equal(name,mr_len(name),"--queue-bytes")==1{flag=2}
   if mr_equal(name,mr_len(name),"--queue-timeout-ms")==1{flag=4}
   if flag==0{break}
   if (options&flag)!=0||route_first+1>=argc{return 1}
   let value:i64=ms_number(argv[route_first+1] as *u8)
   if value<=0{return 1}
   if flag==1{workers=value}
   if flag==2{queue_bytes=value}
   if flag==4{queue_wait_ms=value}
   options=options|flag;route_first=route_first+2
  }
  if mr_config(argc, argv, route_first) != 1 {
      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)
      return 1
  }
  let request_cap: i64 = ms_number(argv[3] as *u8)
  let response_cap: i64 = ms_number(argv[4] as *u8)
  let chunk_cap: i64 = ms_number(argv[5] as *u8)
  let timeout_secs: i64 = ms_number(argv[6] as *u8)
  if timeout_secs < 1 { return 1 }
  if request_cap < 1 { return 1 }
  if response_cap < 1 { return 1 }
  if chunk_cap < 1 { return 1 }
  if chunk_cap > request_cap { return 1 }
  if workers > NX_RA_SIZE_MAX / __size_of(NxMcpPending) { return 1 }
  let default_queue_bytes:i64=mq_default_budget(workers,request_cap)
  if default_queue_bytes<=0{return 1}
  if timeout_secs>NX_RA_SIZE_MAX/1000{return 1}
  // Bootstrap queue envelope derives from configured request memory and time,
  // independently overridable; executor capacity remains unchanged.
  if queue_bytes==0{queue_bytes=default_queue_bytes}
  if queue_wait_ms==0{queue_wait_ms=timeout_secs*1000}
  let queue:*NxMcpQueue=mq_new(queue_bytes,queue_wait_ms)
  if queue==(0 as *NxMcpQueue){return 1}
  // Reserve the protocol pipe before redirecting every library diagnostic to stderr.
  let protocol_fd: i64 = __syscall(NX_SYS_FCNTL, MS_STDOUT, MS_F_DUPFD_CLOEXEC, MS_FIRST_PRIVATE_FD, 0, 0, 0)
  if protocol_fd < MS_FIRST_PRIVATE_FD { return 1 }
  if sys_dup3(MS_STDERR, MS_STDOUT, 0) < 0 { return 1 }
  let body: *u8 = sys_mmap(request_cap)
  let chunk: *u8 = sys_mmap(chunk_cap)
  let control: *NxMcpControl = sys_mmap(__size_of(NxMcpControl)) as *NxMcpControl
  let pool: *NxRequestPool = rp_new(workers,response_cap)
  if pool == (0 as *NxRequestPool) { return 1 }
  let pending: *u8 = sys_mmap(workers*__size_of(NxMcpPending))
  var input_closed:i64=0
  var used: i64 = 0
  var overflow: i64 = 0
  while true {
      if ms_dispatch_queue(queue,pool,pending,protocol_fd,timeout_secs,response_cap,argc,argv,route_first)!=0{rp_cancel_all(pool);return 12}
      if input_closed!=0&&pool.active==0&&queue.count==0{return 0}
      if pool.active>0||queue.count>0{
          let ready:i64=mp_recover_pump_timeout(pool,pending,1-input_closed,protocol_fd,timeout_secs,mq_wait(queue,sys_now_ms()))
          if ready<0{rp_cancel_all(pool);return 12}
          if ready==0{continue}
      }
      if input_closed!=0{continue}
      let n:i64=sys_read(MS_STDIN,chunk,chunk_cap)
      if n==RP_EINTR{continue}
      if n<0{rp_cancel_all(pool);return 2}
      if n==0{
          if used!=0||overflow!=0{rp_cancel_all(pool);return 3}
          input_closed=1;continue
      }
      var i: i64 = 0
      while i < n {
          let c: u8 = chunk[i]
          if overflow != 0 {
              // Consuming an oversized request line to its newline. On the newline, answer the id we
              // retained (if any) with -32600, else a stderr diagnostic, and keep serving.
              if c == (10 as u8) {
                  var okind: i64 = 0
                  var ooff: i64 = 0
                  var olen: i64 = 0
                  if mo_prefix_id(body, used, &okind, &ooff, &olen) == 1 {
                      if mpe_write_error(protocol_fd, okind, ((body as i64) + ooff) as *u8, olen, 0,
                          MSO_CODE_INVALID, "request-line" as *u8,
                          "the request line exceeded the configured byte budget" as *u8, request_cap,
                          "split the request or raise the request byte budget" as *u8) != 0 { rp_cancel_all(pool); return 12 }
                  } else {
                      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)
                  }
                  used = 0
                  overflow = 0
              }
              i = i + 1
              continue
          }
          if c == (10 as u8) {
              if used > 0 {
                  if mct_read(body,used,control) != 0 {
                      // A malformed request line does not end the session: answer -32600 for the id if
                      // one is readable from the raw line, else a stderr diagnostic, then keep serving.
                      var mkind: i64 = 0
                      var moff: i64 = 0
                      var mlen: i64 = 0
                      if mo_prefix_id(body, used, &mkind, &moff, &mlen) == 1 {
                          if mpe_write_error(protocol_fd, mkind, ((body as i64) + moff) as *u8, mlen, 0,
                              MSO_CODE_INVALID, "control-envelope" as *u8,
                              "malformed or ambiguous request metadata" as *u8, used,
                              "correct the JSON-RPC envelope and resend" as *u8) != 0 { rp_cancel_all(pool); return 12 }
                      } else {
                          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)
                      }
                      used = 0
                      i = i + 1
                      continue
                  }
                  if control.method == MCT_CANCEL {
                      var cancelled:i64=mq_cancel(queue,body,control,protocol_fd)
                      if cancelled==1{used=0;i=i+1;continue}
                      if cancelled== -2{rp_cancel_all(pool);return 12}
                      if cancelled==0{cancelled=mp_cancel(pool,pending,body,control)}
                      if cancelled < 0 {
                          // A cancel that cannot be honoured (an initialize, or a reap fault) is not a
                          // reason to end the session. notifications/cancelled has no id, so no response.
                          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)
                      }
                      if cancelled > 0 {
                          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)
                      }
                      used=0; i=i+1; continue
                  }
                  let admitted:i64=mq_enqueue(queue,pool,pending,body,used,control,sys_now_ms(),protocol_fd)
                  if admitted<0{rp_cancel_all(pool);return 12}
                  used = 0
              }
          } else {
              if used >= request_cap { overflow = 1; i = i + 1; continue }
              body[used] = c; used = used + 1
          }
          i = i + 1
      }
  }
  return 0
413}
nx_mcp_stdio.nx source

`nx_mcp_stdio.nx` source
