sessionOf

tcmts.cli.test.e2e.mcp.sessionOf · #3tujbaq4 · term · runs on the server

uses node:child_process

An MCP server run as node <argv>, as an agent's client talks to it. A request the server doesn't answer, in two minutes or before it exits (or one asked once it has), fails with what the server printed to stderr: a test doesn't hang on it. env adds to its environment.

import { plainEnv } from "#codebase/tcmts/cli/test/e2e/harness";
import type { Reply } from "#codebase/tcmts/cli/test/e2e/mcp";
import { spawn } from "node:child_process";
import type { ChildProcess } from "node:child_process";

export function sessionOf(
  dir: string,
  keep: <C extends ChildProcess>(c: C) => C,
  argv: string[],
  env: Record<string, string> = {},
) {
  const child = keep(
    spawn(process.execPath, argv, {
      cwd: dir,
      env: { ...plainEnv(), ...env },
      stdio: ["pipe", "pipe", "pipe"],
    }),
  );
  let buf = "";
  let id = 0;
  const waiting = new Map<number, { resolve: (m: Reply) => void; reject: (e: Error) => void }>();
  let stderr = "";
  child.stderr.on("data", (d) => (stderr += d));
  let exited: string | undefined;
  const gone = () =>
    new Error(
      `the MCP server (${argv.slice(1).join(" ")}) exited (${exited}) without answering:\n${stderr.slice(-4000)}`,
    );
  child.on("exit", (code, signal) => {
    exited = String(signal ?? code);
    for (const w of waiting.values()) w.reject(gone());
    waiting.clear();
  });
  const notes: { method: string; params: Record<string, unknown> }[] = [];
  child.stdout.on("data", (d) => {
    buf += d;
    for (let nl = buf.indexOf("\n"); nl >= 0; nl = buf.indexOf("\n")) {
      const m = JSON.parse(buf.slice(0, nl)) as {
        id?: number;
        method?: string;
        params?: Record<string, unknown>;
      } & Reply;
      buf = buf.slice(nl + 1);
      if (m.id === undefined) notes.push({ method: m.method!, params: m.params! });
      else {
        waiting.get(m.id)?.resolve(m);
        waiting.delete(m.id);
      }
    }
  });
  const request = (method: string, params: Record<string, unknown> = {}) =>
    new Promise<Reply>((resolve, reject) => {
      if (exited !== undefined) return reject(gone());
      const i = ++id;
      const timer = setTimeout(() => {
        waiting.delete(i);
        reject(
          new Error(
            `the MCP server (${argv.slice(1).join(" ")}) didn't answer ${method}${method === "tools/call" ? ` ${String(params.name)}` : ""} in 2 minutes; it printed:\n${stderr.slice(-4000)}`,
          ),
        );
      }, 120_000);
      waiting.set(i, {
        resolve: (m) => (clearTimeout(timer), resolve(m)),
        reject: (e) => (clearTimeout(timer), reject(e)),
      });
      child.stdin.write(JSON.stringify({ jsonrpc: "2.0", id: i, method, params }) + "\n");
    });
  const call = async (
    name: string,
    args: Record<string, unknown> = {},
    meta?: Record<string, unknown>,
  ) => {
    const r = await request("tools/call", {
      name,
      arguments: args,
      ...(meta ? { _meta: meta } : {}),
    });
    const result = r.result as { content: { text: string }[]; isError: boolean };
    return { text: result.content[0]!.text, isError: result.isError };
  };
  return {
    request,
    call,
    notes,
    close: () => child.stdin.end(),
    /** the server's process (the supervisor's, `tcm mcp`) */
    pid: child.pid!,
    kill: (signal: NodeJS.Signals = "SIGKILL") => child.kill(signal),
    /** once it has exited */
    exit: new Promise<void>((resolve) =>
      exited !== undefined ? resolve() : child.once("exit", () => resolve()),
    ),
  };
}

uses

used by