code wiki / (root) / nx_mcp_stdio_http_t174.nx
`nx_mcp_stdio_http_t174.nx` source

↩ module page · 403 lines · 22206 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.nx"
20import "nx_mcp_control.nx"
21import "nx_mcp_pending_http_t174.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 {
  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
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 {
  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
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 {
  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
  } }
  if count < 2 {
      err.code = MSE_CODE_BODY; err.stage = "response-body" as *u8
      err.cause = "the response body is too short to be a JSON object" as *u8; err.measured = count
      err.next = "verify the upstream response is a complete JSON object" as *u8
      return 8
  }
  if out[off] != (123 as u8) {
      err.code = MSE_CODE_BODY; err.stage = "response-body" as *u8
      err.cause = "the response body is not a JSON object" as *u8; err.measured = count
      err.next = "verify the upstream response is a complete JSON object" as *u8
      return 8
  }
  if mr_response_matches(body, body_n, out + off, count) != 1 {
      err.code = MSE_CODE_IDENTITY; err.stage = "response-identity" as *u8
      err.cause = "the response id is absent, ambiguous, invalid or different from the request id" as *u8; err.measured = 0
      err.next = "inspect server request/response correlation; verify write outcome before retrying" as *u8
      return 14
  }
  if mc_write_n(protocol_fd, out + off, count) != 0 {
      err.code = MSE_CODE_PROTOWRITE; err.stage = "protocol-write" as *u8
      err.cause = "writing the response to the protocol channel failed" as *u8; err.measured = 0
      err.next = "the downstream reader closed; reconnect the session" as *u8
      return 9
  }
  if mc_write_n(protocol_fd, "\n" as *u8, 1) != 0 {
      err.code = MSE_CODE_PROTOWRITE; err.stage = "protocol-write" as *u8
      err.cause = "writing the response terminator 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
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 {
  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 MS_FATAL }
  return MS_OK
246}
247
248func 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
253}
254
255func main(argc: i64, argv: *i64) -> i64 {
  var route_first: i64 = 7
  var workers: i64 = 1 // preserve the existing serial invocation contract
  if argc > 7 {
      if mr_equal(argv[7] as *u8,mr_len(argv[7] as *u8),"--workers" as *u8) == 1 {
          if argc < 9 { return 1 }
          workers = ms_number(argv[8] as *u8)
          if workers < 1 { return 1 }
          route_first = 9
      }
  }
  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>] [<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 }
  // 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 first_request: i64 = 1
  var used: i64 = 0
  var overflow: i64 = 0
  while true {
      if pool.active > 0 {
          let ready: i64 = mp_recover_pump(pool,pending,1,protocol_fd,timeout_secs)
          if ready < 0 { rp_cancel_all(pool); return 12 }
          if ready == 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 { rp_cancel_all(pool); return 3 }
          if overflow != 0 { rp_cancel_all(pool); return 3 }
          if ms_drain(pool,pending,protocol_fd,timeout_secs) != 0 { rp_cancel_all(pool); return 12 }
          return 0
      }
      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 {
                      let cancelled: i64 = 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
                  }
                  // A notification cannot be refused and has no response channel; guarantee a free worker
                  // by draining rather than letting admission's capacity refusal end the session.
                  if control.id_kind == 0 {
                      if ms_drain(pool,pending,protocol_fd,timeout_secs) != 0 { rp_cancel_all(pool); return 12 }
                  }
                  let barrier: i64 = first_request != 0 || control.method == MCT_INITIALIZE || control.method == MCT_INITIALIZED
                  if barrier != 0 {
                      if ms_drain(pool,pending,protocol_fd,timeout_secs) != 0 { rp_cancel_all(pool); return 12 }
                  }
                  let admission: i64 = mp_admit(pool,body,control,protocol_fd)
                  // admission -1 now only means a protocol-fd write failed (broken stdout) -- a
                  // notification at capacity was drained above -- so this is genuinely fatal.
                  if admission < 0 {
                      rp_cancel_all(pool); return 17
                  }
                  if admission == 0 { used=0; i=i+1; continue }
                  let rc: i64 = ms_request(argv[1] as *u8, argv[2] as *u8, body, used, response_cap, protocol_fd, timeout_secs, argc, argv, route_first, pool, pending, control)
                  if rc == MS_DUP {
                      if mpe_write_error(protocol_fd, control.id_kind, ((body as i64) + control.id_off) as *u8, control.id_len, 0,
                          MSO_CODE_DUP, "admission" as *u8,
                          "a request with this id is already in flight" as *u8, 0,
                          "use a unique request id or cancel the in-flight request first" as *u8) != 0 { rp_cancel_all(pool); return 12 }
                      used=0; i=i+1; continue
                  }
                  if rc != 0 { mc_puts("NISHI-MCP-STDIO worker could not be created (resource/allocation failure); connection closed\n" as *u8); rp_cancel_all(pool); return rc }
                  if barrier != 0 {
                      if ms_drain(pool,pending,protocol_fd,timeout_secs) != 0 { rp_cancel_all(pool); return 12 }
                  }
                  first_request = 0
                  used = 0
              }
          } else {
              if used >= request_cap { overflow = 1; i = i + 1; continue }
              body[used] = c; used = used + 1
          }
          i = i + 1
      }
  }
  return 0
403}
nx_mcp_stdio_http_t174.nx source

`nx_mcp_stdio_http_t174.nx` source
