///|
/// StdioClientTransport — js implementation (node + bun, one code path).
///
/// MCP host/client connecting to a local MCP server via spawned child
/// process, communicating with newline-delimited JSON-RPC over stdin/stdout
/// pipes; stderr is inherited for server-side logging. Lifecycle mirrors the
/// native implementation in stdio_client.mbt (two-phase start, graceful
/// close = stdin EOF, forced kill after timeout).
///
/// Bridge architecture (see scripts probe conclusions): the moon js backend
/// cannot take `node:stream`/`node:child_process` members as values —
/// `#module` RHS forms are callee-only and dotted member paths generate
/// invalid ESM — so all Node interop lives in inline `extern "js"` snippets
/// that dynamically `import('node:...')` and return Promises; the Promises
/// are bridged into the coroutine world with `@js_async.Promise::wait`:
/// - read: child.stdout → `Readable.toWeb` → `@js_async.ReadableStream::
/// from_js` (pipe intermediary with cancellation) → buffered
/// `read_until("\n")`, exactly like the native frame handling;
/// - write: child.stdin → `Writable.toWeb` → default writer; each
/// `send` is one awaited `write` (message + newline) so backpressure
/// and "written before send returns" are preserved;
/// - exit: the exit listener is registered inside the spawn extern (the
/// child may exit the moment stdin EOF arrives), and every promise the
/// MoonBit side waits on is guaranteed to settle;
/// - close: stdin EOF + a JS-side kill timer, disarmed by the exit
/// promise — the sync `close()` cannot hang;
/// - cancel: a watchdog task spawned into the caller's TaskGroup reruns
/// the same choreography when the group is cancelled/fails, so the
/// child never outlives the group (native parity via the native
/// transport's group-attached `cancel_handler`).
///|
/// Opaque JavaScript value (spawn options object, node stream handles).
#external
priv type JsAny
///|
/// Spawned child process (opaque node ChildProcess handle).
#external
priv type JsChildProcess
///|
/// Default writer of the web-converted child stdin stream (opaque).
#external
priv type JsStdinWriter
///|
/// Plain JS object `{ code : Int }` resolved by the exit listener
/// (`code = -1` when the child was killed by a signal — node passes null).
priv struct ChildExit {
code : Int
}
///|
/// Result of a successful spawn: the child plus its pre-registered exit
/// promise. The extern shapes the plain JS object `{ child, exit }`.
priv struct StartedChild {
child : JsChildProcess
exit : @js_async.Promise[ChildExit]
}
///|
/// Spawn `cmd`, resolving with the child and its exit promise on the
/// node 'spawn' event, rejecting on spawn failure ('error' event).
/// Both listeners attach synchronously at spawn time because node emits
/// child 'error' on nextTick — before any promise continuation could.
/// A permanent no-op 'error' handler is installed on success so late
/// 'error' events (e.g. kill failures) cannot crash the process.
extern "js" fn js_start_child(
cmd : String,
args : Array[String],
options : JsAny,
) -> @js_async.Promise[StartedChild] =
#| async (cmd, args, options) => {
#| const { spawn } = await import('node:child_process')
#| return new Promise((resolve, reject) => {
#| const child = spawn(cmd, args, options)
#| const exit = new Promise((res) => {
#| child.once('exit', (code) => res({ code: code === null ? -1 : code }))
#| })
#| child.once('spawn', () => {
#| child.on('error', () => {})
#| resolve({ child, exit })
#| })
#| child.once('error', (err) => reject(err))
#| })
#| }
///|
extern "js" fn child_prop(child : JsChildProcess, name : String) -> JsAny =
#| (c, k) => c[k]
///|
extern "js" fn js_kill(child : JsChildProcess) -> Unit =
#| (c) => {
#| try { c.kill() } catch {}
#| }
///|
/// Spawn options: stdin/stdout piped, stderr inherited (server-side
/// logging, per the MCP spec), environment = parent env + `extra_env`.
extern "js" fn spawn_options(
env_keys : Array[String],
env_vals : Array[String],
) -> JsAny =
#| (keys, vals) => {
#| const env = { ...globalThis.process.env }
#| for (let i = 0; i < keys.length; i++) env[keys[i]] = vals[i]
#| return { stdio: ['pipe', 'pipe', 'inherit'], env }
#| }
///|
extern "js" fn readable_to_web(
stream : JsAny,
) -> @js_async.Promise[@js_async.JsReadableStream] =
#| async (stream) => {
#| const { Readable } = await import('node:stream')
#| return Readable.toWeb(stream)
#| }
///|
extern "js" fn writable_to_web(stream : JsAny) -> @js_async.Promise[JsAny] =
#| async (stream) => {
#| const { Writable } = await import('node:stream')
#| return Writable.toWeb(stream)
#| }
///|
extern "js" fn stream_writer(ws : JsAny) -> JsStdinWriter =
#| (ws) => ws.getWriter()
///|
/// One awaited write per `send`/`send_notification` (message + newline).
extern "js" fn writer_write(
writer : JsStdinWriter,
data : String,
) -> @js_async.Promise[Unit] =
#| (w, s) => w.write(s)
///|
/// Fire-and-forget graceful close: end stdin (EOF signal), arm a kill
/// timer disarmed by the pre-registered exit promise. Runs entirely on the
/// JS side so the sync MoonBit `close()` cannot hang; writer-close
/// rejections (child already gone) are swallowed.
extern "js" fn graceful_close(
writer : JsStdinWriter,
child : JsChildProcess,
exit : @js_async.Promise[ChildExit],
) -> Unit =
#| (writer, child, exit) => {
#| const kill_timer = setTimeout(() => {
#| try { child.kill() } catch {}
#| }, 5000)
#| exit.then(() => clearTimeout(kill_timer)).catch(() => clearTimeout(kill_timer))
#| try { writer.close().catch(() => {}) } catch {}
#| }
///|
pub struct StdioClientTransport {
cmd : String
args : Array[String]
extra_env : Map[String, String]
mut closed : Bool
/// Buffered stdout reader; `Some` marks the transport as started.
priv mut reader : &@io.Reader?
/// Cancels the from_js copy coroutine and the underlying JS stream.
priv mut stop_reader : () -> Unit
priv mut child : JsChildProcess?
priv mut writer : JsStdinWriter?
priv mut exit : @js_async.Promise[ChildExit]?
}
///|
pub fn StdioClientTransport::StdioClientTransport(
cmd~ : String,
args? : Array[String] = [],
extra_env? : Map[String, String] = {},
) -> StdioClientTransport {
{
cmd,
args,
extra_env,
closed: false,
reader: None,
stop_reader: fn() { () },
child: None,
writer: None,
exit: None,
}
}
///|
/// Spawn the child process inside the given TaskGroup and build the
/// stdin/stdout bridges. Must be called before `send()` / `receive()`.
///
/// A watchdog task is spawned into `group` after a successful spawn, tying
/// the child's lifetime to the group (native parity: `@process.spawn` there
/// attaches `graceful_cancel` to the group). The watchdog waits on the
/// pre-registered exit promise in a cancellable suspension; when the group
/// tears down — cancellation, whole-group failure, or body return without a
/// manual `close()` — it mirrors close()'s JS choreography and only then
/// lets the group finish, so `with_task_group` never returns with a leaked
/// child.
pub async fn StdioClientTransport::start(
self : StdioClientTransport,
group : @async.TaskGroup[Unit],
) -> Unit raise @types.TransportError {
if self.closed {
raise @types.InvalidState("Cannot start a closed transport")
}
if self.reader is Some(_) {
raise @types.InvalidState("Transport already started")
}
let started = js_start_child(
self.cmd,
self.args,
spawn_options(env_keys(self.extra_env), env_vals(self.extra_env)),
).wait() catch {
e =>
raise @types.WriteError(
"Failed to spawn process '\{self.cmd}': " + e.to_string(),
)
}
// Fallible bridging first; on failure the child is killed so a partial
// start never leaks a process.
let stdout_web = readable_to_web(child_prop(started.child, "stdout")).wait() catch {
e => {
js_kill(started.child)
raise @types.ReadError("Failed to bridge child stdout: " + e.to_string())
}
}
let stdin_web = writable_to_web(child_prop(started.child, "stdin")).wait() catch {
e => {
js_kill(started.child)
raise @types.WriteError("Failed to bridge child stdin: " + e.to_string())
}
}
let rs = @js_async.ReadableStream::from_js(stdout_web)
let reader : &@io.Reader = rs
let writer = stream_writer(stdin_web)
self.reader = Some(reader)
self.stop_reader = fn() { rs.close() }
self.child = Some(started.child)
self.writer = Some(writer)
self.exit = Some(started.exit)
// Watchdog: see watch_child_in_group. `no_wait` matters: the group's
// waiting counter must not include the watchdog, so that when the group
// body returns without a manual close(), the group still cancels it at
// teardown (same as the native group-attached cancel_handler) instead of
// waiting forever for a child that nobody asked to exit.
group.spawn_bg(no_wait=true, () => {
watch_child_in_group(writer, started.child, started.exit)
})
}
///|
/// TaskGroup watchdog — js counterpart of the native transport attaching
/// `@process.graceful_cancel` to the group. Spawned after a successful
/// `start`, it keeps the child's lifetime inside the group:
///
/// - Normal path: cancellable wait on the pre-registered exit promise; the
/// child exits on its own (any code — exit-code semantics belong to
/// receive()'s EOF handling) and the task ends without failing the group.
/// - Cancelled path (group cancellation, whole-group failure, or teardown
/// with the body returned without a manual close): rerun close()'s
/// JS choreography (stdin EOF + 5s kill timer, guaranteeing the exit
/// promise settles), then block on the exit promise uncancellably — the
/// group must not tear down while the child may still be alive, matching
/// the native SIGTERM → 5s → SIGKILL blocking wait — and finally re-raise
/// the cancellation (a cancelled task is not a group failure).
/// - Racing a manual `close()` is safe: JS-side writer.close()/kill are
/// idempotent, and if close() already ended stdin the wait resolves
/// instead of hanging.
///
/// The catch is a cancellation handler that performs async work, not mere
/// cleanup, so the errdefer rewrite suggested by fragile_catch_all cannot
/// express it (same shape as with_task_group's own teardown).
#warnings("-fragile_catch_all")
async fn watch_child_in_group(
writer : JsStdinWriter,
child : JsChildProcess,
exit : @js_async.Promise[ChildExit],
) -> Unit {
// The AbortController is never observed by the exit promise itself; it
// only makes the suspension yield to group cancellation.
let ctrl = @js_async.AbortController::new()
let _ = exit.wait(abort_controller=ctrl) catch {
err => {
graceful_close(writer, child, exit)
try {
let _ = exit.wait()
} catch {
// The exit promise never rejects by construction; swallow anyway so
// the re-raise below carries the cancellation, not a JsError (a
// non-Cancelled error here would fail the whole group).
_ => ()
}
raise err
}
}
}
///|
/// Flatten a Map for the JS env-object builder (Map iteration preserves
/// insertion order, so keys and values stay aligned).
fn env_keys(extra : Map[String, String]) -> Array[String] {
let keys = []
for k, _ in extra {
keys.push(k)
}
keys
}
///|
fn env_vals(extra : Map[String, String]) -> Array[String] {
let vals = []
for _, v in extra {
vals.push(v)
}
vals
}
///|
/// Read one JSON-RPC message from the child's stdout.
/// Returns None on EOF (child closed stdout / process exited).
pub async fn StdioClientTransport::receive(
self : StdioClientTransport,
) -> String? raise @types.TransportError {
if self.closed {
return None
}
let reader = match self.reader {
Some(r) => r
None =>
raise @types.InvalidState("Transport not started — call start() first")
}
let line_opt = reader.read_until("\n") catch {
e =>
raise @types.ReadError(
"Failed to read from child process: " + e.to_string(),
)
}
match line_opt {
None => {
self.closed = true
None
}
Some(line) => {
let message = line.trim().to_owned()
if message.is_empty() {
return self.receive()
}
Some(message)
}
}
}
///|
/// Send a JSON-RPC message to the child's stdin. One awaited write carries
/// message + newline — the same wire bytes as the native buffered write +
/// flush, with backpressure preserved.
pub async fn StdioClientTransport::send(
self : StdioClientTransport,
message : String,
) -> Unit raise @types.TransportError {
if self.closed {
raise @types.InvalidState("Cannot send on closed transport")
}
let writer = match self.writer {
Some(w) => w
None =>
raise @types.InvalidState("Transport not started — call start() first")
}
match validate_jsonrpc_message(message) {
Err(e) =>
raise @types.WriteError("Invalid JSON-RPC message: " + e.message())
Ok(_) => ()
}
writer_write(writer, message + "\n").wait() catch {
e =>
raise @types.WriteError(
"Failed to write to child stdin: " + e.to_string(),
)
}
}
///|
/// Send a JSON-RPC notification (no id, fire-and-forget) to the child's stdin.
pub async fn StdioClientTransport::send_notification(
self : StdioClientTransport,
notification : @types.Notification,
) -> Unit raise @types.TransportError {
if self.closed {
raise @types.InvalidState("Cannot send on closed transport")
}
let writer = match self.writer {
Some(w) => w
None =>
raise @types.InvalidState("Transport not started — call start() first")
}
let json = notification.to_jsonrpc_string()
writer_write(writer, json + "\n").wait() catch {
e =>
raise @types.WriteError("Failed to send notification: " + e.to_string())
}
}
///|
/// Graceful shutdown per MCP spec:
/// 1. Close stdin (EOF signal to the child) via the JS-side choreography,
/// with a 5s kill timer disarmed by the child's exit.
/// 2. Cancel the stdout reader bridge.
/// 3. Mark the transport as closed.
pub fn StdioClientTransport::close(self : StdioClientTransport) -> Unit {
if self.closed {
return
}
self.closed = true
match self.writer {
Some(writer) =>
match (self.child, self.exit) {
(Some(child), Some(exit)) => graceful_close(writer, child, exit)
_ => ()
}
None =>
// Partial start (or never started): no stdin to close politely.
match self.child {
Some(child) => js_kill(child)
None => ()
}
}
self.writer = None
self.child = None
self.exit = None
(self.stop_reader)()
self.reader = None
}