mesh-sdk/test/stdio.test.ts

// A tools bundle as a process (novox/hq ADR 0188): the stdio loop answers initialize, tools/list and
// tools/call the way the node's runtime asks, and a tool that throws is a tool error, not a hang-up.
import assert from "node:assert/strict";
import { test } from "node:test";
import { spawn } from "node:child_process";
import { fileURLToPath } from "node:url";

test("a bundle on stdio lists its tools and answers a call, a seat's verb among them", async () => {
  const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-bundle.mjs", import.meta.url))], { stdio: ["pipe", "pipe", "inherit"] });
  const replies: any[] = [];
  let buffered = "";
  child.stdout.on("data", (d) => {
    buffered += d.toString();
    let at: number;
    while ((at = buffered.indexOf("\n")) >= 0) {
      replies.push(JSON.parse(buffered.slice(0, at)));
      buffered = buffered.slice(at + 1);
    }
  });
  const ask = async (id: number, method: string, params?: unknown) => {
    child.stdin.write(JSON.stringify({ jsonrpc: "2.0", id, method, params }) + "\n");
    for (let i = 0; i < 100 && !replies.some((r) => r.id === id); i++) await new Promise((r) => setTimeout(r, 20));
    return replies.find((r) => r.id === id);
  };
  try {
    assert.equal((await ask(1, "initialize", { protocolVersion: "2025-03-26", capabilities: {} })).result.serverInfo.name, "lamp");
    child.stdin.write(JSON.stringify({ jsonrpc: "2.0", method: "notifications/initialized" }) + "\n");
    const listed = (await ask(2, "tools/list")).result.tools;
    assert.deepEqual(listed.map((t: any) => t.name), ["brightness", "node-lamp.on", "fail"]);
    assert.deepEqual(listed[0].inputSchema, { type: "object", properties: { level: { type: "number", description: "0 to 10" } } });
    const called = (await ask(3, "tools/call", { name: "brightness", arguments: { level: 7 } })).result;
    assert.deepEqual(JSON.parse(called.content[0].text), { level: 7 });
    const failed = (await ask(4, "tools/call", { name: "fail", arguments: {} })).result;
    assert.equal(failed.isError, true);
    assert.equal(failed.content[0].text, "the bulb is gone");
    assert.equal((await ask(5, "tools/call", { name: "nothing" })).error.code, -32602);
  } finally {
    child.stdin.end();
    child.kill();
  }
});

test("a bundle served as its module lists the seat's verbs as the seat's, whichever registered first (ADR 0193)", async () => {
  const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-seat-first.mjs", import.meta.url))],
    { stdio: ["pipe", "pipe", "inherit"], env: { ...process.env, MESH_SERVED_MODULE: "nftables", NFT_WORD: "given" } });
  const replies: any[] = [];
  let buffered = "";
  child.stdout.on("data", (d) => {
    buffered += d.toString();
    let at: number;
    while ((at = buffered.indexOf("\n")) >= 0) {
      replies.push(JSON.parse(buffered.slice(0, at)));
      buffered = buffered.slice(at + 1);
    }
  });
  const ask = async (id: number, method: string, params?: unknown) => {
    child.stdin.write(JSON.stringify({ jsonrpc: "2.0", id, method, params }) + "\n");
    for (let i = 0; i < 100 && !replies.some((r) => r.id === id); i++) await new Promise((r) => setTimeout(r, 20));
    return replies.find((r) => r.id === id);
  };
  try {
    assert.equal((await ask(1, "initialize", { protocolVersion: "2025-03-26", capabilities: {} })).result.serverInfo.name, "nftables");
    const listed = (await ask(2, "tools/list")).result.tools;
    assert.deepEqual(listed.map((t: any) => t.name), ["node-packet-filter.rules", "firewall_rules"]);
    const called = (await ask(3, "tools/call", { name: "firewall_rules", arguments: {} })).result;
    assert.equal(called.content[0].text, JSON.stringify("given"));
  } finally {
    child.kill();
  }
});

test("a launched tool's emit is asked of the runtime, and the call answers once the runtime has published (ADR 0193)", async () => {
  const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-emits.mjs", import.meta.url))],
    { stdio: ["pipe", "pipe", "inherit"], env: { ...process.env, MESH_SERVED_MODULE: "chime", MESH_MODULE: "chime", MESH_NODE: "anchor" } });
  const seen: any[] = [];
  let buffered = "";
  child.stdout.on("data", (d) => {
    buffered += d.toString();
    let at: number;
    while ((at = buffered.indexOf("\n")) >= 0) {
      seen.push(JSON.parse(buffered.slice(0, at)));
      buffered = buffered.slice(at + 1);
    }
  });
  const until = async (match: (m: any) => boolean) => {
    for (let i = 0; i < 150 && !seen.some(match); i++) await new Promise((r) => setTimeout(r, 20));
    return seen.find(match);
  };
  const send = (m: unknown) => child.stdin.write(JSON.stringify(m) + "\n");
  try {
    send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {} } });
    await until((m) => m.id === 1);
    send({ jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "ring", arguments: {} } });
    const asked = await until((m) => m.method === "mesh/publish");
    assert.ok(asked, "the tool's emit was not asked of the runtime");
    assert.equal(asked.params.key, "rang");
    assert.deepEqual(asked.params.body, { times: 1 });
    assert.equal(asked.params.headers["x-source"], "chime");
    assert.equal(seen.find((m) => m.id === 2), undefined, "the call answered before its emit was published");
    send({ jsonrpc: "2.0", id: asked.id, result: {} });
    const answered = await until((m) => m.id === 2);
    assert.deepEqual(JSON.parse(answered.result.content[0].text), { rang: true });
  } finally {
    child.kill();
  }
});

test("a launched bundle subscribes as it is imported, handles events the runtime hands it, and asks through it (ADR 0198)", async () => {
  const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-long-running.mjs", import.meta.url))],
    { stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, MESH_SERVED_MODULE: "watcher" } });
  const seen: any[] = [];
  let buffered = "";
  let said = "";
  child.stderr.on("data", (d) => (said += d.toString()));
  child.stdout.on("data", (d) => {
    buffered += d.toString();
    let at: number;
    while ((at = buffered.indexOf("\n")) >= 0) {
      seen.push(JSON.parse(buffered.slice(0, at)));
      buffered = buffered.slice(at + 1);
    }
  });
  const until = async (match: (m: any) => boolean) => {
    for (let i = 0; i < 150 && !seen.some(match); i++) await new Promise((r) => setTimeout(r, 20));
    return seen.find(match);
  };
  const send = (m: unknown) => child.stdin.write(JSON.stringify(m) + "\n");
  const event = (id: number, key: string, body: unknown) =>
    send({ jsonrpc: "2.0", id, method: "mesh/event", params: { envelope: { key, node: "anchor", body, headers: { "x-event-id": `e${id}`, "x-source": "alpha" } } } });
  try {
    // The subscription comes first, from the import, before the bundle has said what it serves.
    const sub = await until((m) => m.method === "mesh/subscribe");
    assert.deepEqual(sub.params, { pattern: "alpha.happened" });
    send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {} } });
    assert.equal(await until((m) => m.id === 1), undefined, "initialize was answered while the import still waited on its subscription");
    send({ jsonrpc: "2.0", id: sub.id, result: {} });
    assert.equal((await until((m) => m.id === 1)).result.serverInfo.name, "watcher");
    send({ jsonrpc: "2.0", id: 2, method: "tools/list" });
    assert.deepEqual((await until((m) => m.id === 2)).result.tools.map((t: any) => t.name), ["relay"]);

    // Handled: answered with a result, so the runtime acknowledges it.
    event(10, "alpha.happened", { n: 1 });
    assert.deepEqual((await until((m) => m.id === 10)).result, {});
    assert.match(said, /handled alpha\.happened \{"n":1\}/);
    // The handler failed: answered with an error, so the runtime does not acknowledge it.
    event(11, "alpha.happened", { fail: true });
    assert.match((await until((m) => m.id === 11)).error.message, /not this one/);
    // No handler is for it: answered at once, so it is acknowledged rather than redelivered.
    event(12, "beta.other", {});
    assert.deepEqual((await until((m) => m.id === 12)).result, {});

    // A tool asks the bus: the runtime is asked, and its answer is the tool's.
    send({ jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "relay", arguments: {} } });
    const asked = await until((m) => m.method === "mesh/ask");
    assert.deepEqual(asked.params, { key: "beta.one", body: { x: 1 } });
    send({ jsonrpc: "2.0", id: asked.id, result: { beta: 1 } });
    assert.deepEqual(JSON.parse((await until((m) => m.id === 3)).result.content[0].text), { beta: 1 });
  } finally {
    child.kill();
  }
});

test("topics match as the runtime's do: * one segment, # and ** any number", async () => {
  const { topicMatches } = await import("../dist/stdio/index.js");
  assert.ok(topicMatches("alpha.happened", "alpha.happened"));
  assert.ok(topicMatches("alpha.*", "alpha.happened"));
  assert.ok(!topicMatches("alpha.*", "alpha.happened.twice"));
  assert.ok(topicMatches("#", "anything.at.all"));
  assert.ok(topicMatches("alpha.**", "alpha.a.b"));
  assert.ok(!topicMatches("beta.*", "alpha.happened"));
});