242 lines
14 KiB
TypeScript
242 lines
14 KiB
TypeScript
// 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("a launched bundle watches its state as it is imported, has the current values before it goes on, and keeps state through the runtime (ADR 0202)", async () => {
|
|
const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-state.mjs", import.meta.url))],
|
|
{ stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, MESH_SERVED_MODULE: "keeper" } });
|
|
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 change = (id: number, watch: string, c: Record<string, unknown>) =>
|
|
send({ jsonrpc: "2.0", id, method: "mesh/state", params: { watch, change: { state: "servers", revision: id, ...c } } });
|
|
try {
|
|
// The watch comes first, from the import; the current values arrive before it is answered.
|
|
const watch = await until((m) => m.method === "mesh/state.watch");
|
|
assert.deepEqual(watch.params, { state: "servers", key: "" });
|
|
change(100, watch.id, { key: "all.one", op: "put", value: { url: "http://one" }, current: true });
|
|
assert.deepEqual((await until((m) => m.id === 100)).result, {});
|
|
assert.match(said, /was put all\.one \{"url":"http:\/\/one"\}/);
|
|
assert.doesNotMatch(said, /watching/, "the import went on before the watch was answered");
|
|
send({ jsonrpc: "2.0", id: watch.id, result: {} });
|
|
await until(() => said.includes("watching"));
|
|
assert.match(said, /watching/);
|
|
|
|
// A change for a watch this bundle never asked for is refused, not handed to somebody else.
|
|
send({ jsonrpc: "2.0", id: 101, method: "mesh/state", params: { watch: "bundle-999", change: { key: "x", op: "put" } } });
|
|
assert.match((await until((m) => m.id === 101)).error.message, /names no watch/);
|
|
// A change the handler fails is answered as an error.
|
|
change(102, watch.id, { key: "all.two", op: "put", value: { fail: true }, current: false });
|
|
assert.match((await until((m) => m.id === 102)).error.message, /not this one/);
|
|
change(103, watch.id, { key: "all.one", op: "delete", current: false });
|
|
assert.deepEqual((await until((m) => m.id === 103)).result, {});
|
|
assert.match(said, /now delete all\.one null/);
|
|
|
|
send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {} } });
|
|
assert.equal((await until((m) => m.id === 1)).result.serverInfo.name, "keeper");
|
|
|
|
// A tool writes: the runtime is asked, and the revision it answers is the tool's.
|
|
send({ jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "remember", arguments: { key: "all.three", value: { url: "http://three" } } } });
|
|
const put = await until((m) => m.method === "mesh/state.put");
|
|
assert.deepEqual(put.params, { state: "servers", key: "all.three", value: { url: "http://three" } });
|
|
send({ jsonrpc: "2.0", id: put.id, result: { revision: 7 } });
|
|
assert.equal(JSON.parse((await until((m) => m.id === 2)).result.content[0].text), 7);
|
|
// A read of a key with no value is null; a refusal is the tool's error, in the runtime's words.
|
|
send({ jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "recall", arguments: { key: "nothing" } } });
|
|
const get = await until((m) => m.method === "mesh/state.get");
|
|
send({ jsonrpc: "2.0", id: get.id, result: null });
|
|
assert.equal(JSON.parse((await until((m) => m.id === 3)).result.content[0].text), null);
|
|
send({ jsonrpc: "2.0", id: 4, method: "tools/call", params: { name: "list", arguments: {} } });
|
|
const keys = await until((m) => m.method === "mesh/state.keys");
|
|
assert.deepEqual(keys.params, { state: "licence-manager.bindings" });
|
|
send({ jsonrpc: "2.0", id: keys.id, error: { code: -32000, message: "keeper keeps and reads no state called" } });
|
|
const refused = await until((m) => m.id === 4);
|
|
assert.equal(refused.result.isError, true);
|
|
assert.match(refused.result.content[0].text, /keeps and reads no state called/);
|
|
} 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"));
|
|
});
|