mcpWorker

tcmts.cli.mcp.mcpWorker · #kz6eiszi · term · runs on the server

uses node:child_process

The MCP server's worker: one tcm (bin internal-worker, in root) kept running, so a tool's command doesn't start one. It runs one command at a time; run gives undefined if it's busy (run that one in a tcm of its own), and a command whose worker exits under it is undefined too. After recycle commands it's started afresh, so nothing a command leaves behind piles up.

import { plain } from "#codebase/tcmts/cli/mcp";
import { spawn } from "node:child_process";
import type { ChildProcess } from "node:child_process";

export function mcpWorker(bin: string[], root: string, recycle = 100) {
  type Job = {
    id: number;
    onLine?: (line: string) => void;
    resolve: (r: { code: number; out: string } | undefined) => void;
    out: string;
    heard: number;
    done: { out?: boolean; err?: boolean };
    code: number;
  };
  let child: ChildProcess | undefined;
  let runs = 0;
  let job: Job | undefined;
  let ids = 0;
  const start = () => {
    const c = spawn(bin[0]!, [...bin.slice(1), "internal-worker"], {
      cwd: root,
      env: { ...process.env, NO_COLOR: "1", FORCE_COLOR: "0", TCM_BRIEF: "1" },
      stdio: ["pipe", "pipe", "pipe"],
    });
    const pending = { out: "", err: "" };
    const add = (j: Job, text: string) => {
      j.out += text;
      if (!j.onLine) return;
      const end = j.out.lastIndexOf("\n");
      if (end < j.heard) return;
      for (const line of plain(j.out.slice(j.heard, end)).split("\n"))
        if (line.trim()) j.onLine(line);
      j.heard = end + 1;
    };
    const feed = (which: "out" | "err", d: string) => {
      pending[which] += d;
      for (;;) {
        const j = job;
        const z = pending[which].indexOf("\0");
        if (!j) return void (pending[which] = "");
        if (z < 0) return void (add(j, pending[which]), (pending[which] = ""));
        add(j, pending[which].slice(0, z));
        pending[which] = pending[which].slice(z);
        const nl = pending[which].indexOf("\n");
        if (nl < 0) return;
        const [, id, code] = pending[which].slice(1, nl).split(" ");
        pending[which] = pending[which].slice(nl + 1);
        if (Number(id) !== j.id) continue;
        j.done[which] = true;
        if (which === "out") j.code = Number(code);
        if (j.done.out && j.done.err) {
          job = undefined;
          runs++;
          j.resolve({ code: j.code, out: plain(j.out).trim() });
        }
      }
    };
    c.stdout!.setEncoding("utf8").on("data", (d: string) => feed("out", d));
    c.stderr!.setEncoding("utf8").on("data", (d: string) => feed("err", d));
    c.on("exit", () => {
      if (child === c) child = undefined;
      const j = job;
      job = undefined;
      j?.resolve(undefined);
    });
    child = c;
    runs = 0;
  };
  return {
    run(
      argv: string[],
      env: Record<string, string> = {},
      onLine?: (line: string) => void,
    ): Promise<{ code: number; out: string } | undefined> | undefined {
      if (job) return undefined;
      if (child && runs >= recycle) {
        child.stdin!.end();
        child = undefined;
      }
      if (!child) start();
      return new Promise((resolve) => {
        job = { id: ++ids, onLine, resolve, out: "", heard: 0, done: {}, code: 1 };
        child!.stdin!.write(JSON.stringify({ id: job.id, argv, env }) + "\n");
      });
    },
    close() {
      child?.stdin!.end();
      child = undefined;
    },
  };
}

uses

used by