A person's own client, and pins that the wire did not change #13

Merged
jschoubben merged 6 commits from feat/nats-genesis into main 2026-09-27 17:20:07 +00:00
6 changed files with 642 additions and 2 deletions
Showing only changes of commit 9acc40145a - Show all commits
+4 -2
View File
@@ -4,11 +4,13 @@
"description": "The Novox Mesh tool runtime \u2014 binds the mesh broker and serves the assigned modules' tools.",
"type": "module",
"bin": {
"mesh-tools": "./dist/main.js"
"mesh-tools": "./dist/main.js",
"mesh": "./dist/mesh.js"
},
"scripts": {
"build": "tsc",
"test": "node --test --experimental-strip-types 'test/*.test.ts'"
"pretest": "tsc",
"test": "node --test --test-concurrency=1 --experimental-strip-types 'test/*.test.ts'"
},
"dependencies": {
"@novox/mesh-sdk": "^0.1.0",
+125
View File
@@ -0,0 +1,125 @@
/**
* A person's client: the mesh's tools from a workstation (novox/hq design 25 §7).
*
* Two surfaces over one thing. A command line, for somebody at a terminal; an MCP server, for an
* agent. Both are adapters over the same three calls — what tools are there, what does this one take,
* call it — because a second way of reaching a tool is a second thing to keep correct.
*
* **It uses the same client a module's runtime uses.** Not a second protocol and not a bridge: a
* person connects as their own bus user, publishes on the tool subjects their account permits, and the
* server refuses anything else. So "what may this person do" is answered by the same permission list
* that answers it for a module, and there is nothing here for an audit to read separately.
*
* What a person may NOT do is the more interesting half, and none of it is enforced here — it is the
* account (design 25 §4): they cannot publish an event, so they cannot claim a module said something;
* they have no consumer, so there is no delivery to acknowledge; and they cannot answer a request, so
* they cannot impersonate a module on a bus where anyone may serve a tool.
*/
import { readFile } from "node:fs/promises";
import type { Broker } from "@novox/mesh-sdk/messaging";
import { connectNats, type Credential } from "./broker-nats.js";
/** Where the catalogue answers what tools the mesh has. */
const CATALOGUE_TOOLS = "mesh-catalog.catalog_tools";
/** A tool as the catalogue describes one. */
export interface Tool {
module: string;
name: string;
description?: string;
/** The JSON schema of what it takes, as the module declared it. */
input?: unknown;
}
/**
* A person's credential, as `operator issue` prints it.
*
* The same shape a module is handed, minus the parts a module needs and a person does not: no node,
* because a person is not on a machine, and no module, because they are not one.
*/
export interface PersonCredential extends Credential {
person?: string;
invokes?: string[];
}
/** Read the credential from the file `operator issue` produced. */
export async function credentialFrom(path: string): Promise<PersonCredential> {
const raw = await readFile(path, "utf8");
let held: PersonCredential;
try {
held = JSON.parse(raw) as PersonCredential;
} catch (e) {
throw new Error(
`${path} is not a credential this mesh issued: ${(e as Error).message}. ` +
"It is the JSON `operator issue` printed, saved verbatim.",
);
}
if (!held.url || !held.user || !held.password) {
throw new Error(
`${path} names no bus, user or password. It is the JSON \`operator issue\` printed, saved ` +
"verbatim — not an edited copy of it.",
);
}
return held;
}
/** Connect as this person. The module name the runtime wants is their own user, because every subject
* it derives is for a tool somebody else serves. */
export async function connectAs(held: PersonCredential): Promise<Broker> {
return connectNats({ ...held, module: held.user });
}
/**
* What tools the mesh has, asked of the catalogue.
*
* **Asked, not configured.** The catalogue is the only thing that knows what is installed, and a
* client carrying its own list would be a list that goes stale the first time a module is assigned —
* silently, because a tool that is not offered looks exactly like a tool that does not exist.
*/
export async function toolsOn(bus: Broker): Promise<Tool[]> {
const answered = await bus.request<Record<string, never>, { tools?: Tool[] } | Tool[]>(
CATALOGUE_TOOLS,
{},
);
const tools = Array.isArray(answered) ? answered : (answered.tools ?? []);
return tools
.slice()
.sort((a: Tool, b: Tool) => `${a.module}.${a.name}`.localeCompare(`${b.module}.${b.name}`));
}
/** Call one tool. The key is `<module>.<tool>`, which is what a person types and what their account
* permits — one vocabulary, so a refusal names the thing they asked for. */
export async function callTool(bus: Broker, key: string, args: unknown): Promise<unknown> {
if (!key.includes(".")) {
throw new Error(
`"${key}" does not name a tool: write <module>.<tool>, as \`mesh tools\` lists them`,
);
}
return bus.request<unknown, unknown>(key, args ?? {});
}
/**
* Why a call failed, said so that the remedy is in the words.
*
* Three answers a person actually gets, and they need different things done: nobody serves that tool,
* the mesh refused this person, or the tool itself failed. Without this they are one timeout and a
* stack trace.
*/
export function whyItFailed(key: string, err: unknown): string {
const message = err instanceof Error ? err.message : String(err);
if (/no responders|503/i.test(message)) {
return `nothing serves ${key}. The module may not be assigned to any machine, or it is down — ` +
"`mesh tools` lists what the catalogue says is there.";
}
if (/permissions violation|authorization/i.test(message)) {
return `this credential may not call ${key}. What it may call was fixed when it was issued; ` +
"`operator issue` again with the tool named, or ask somebody who can.";
}
if (/timeout/i.test(message)) {
return `${key} did not answer in time. Something is serving it, so this is the tool being slow ` +
"rather than absent.";
}
return `${key} failed: ${message}`;
}
+139
View File
@@ -0,0 +1,139 @@
/**
* The mesh's tools as an MCP server, over stdio (novox/hq design 25 §7).
*
* **A thin adapter and nothing more.** Every tool an agent sees is one the catalogue listed and one
* this credential may call; the schema is the module's own; the answer is the module's own. Nothing
* here decides anything, which is why it is short — an MCP surface that reshaped arguments or
* summarised answers would be a second definition of what a tool is, and the module's manifest is the
* first.
*
* Implemented against the protocol directly rather than through a library: the surface is three
* methods and one framing, and a dependency here would be a dependency on every workstation.
*/
import type { Broker } from "@novox/mesh-sdk/messaging";
import { callTool, toolsOn, whyItFailed, type Tool } from "./client.js";
/** The protocol version this speaks. Stated, because a host that wants another should be told so
* rather than discovering it through a shape it did not expect. */
const PROTOCOL = "2024-11-05";
interface Request {
jsonrpc: string;
id?: number | string | null;
method: string;
params?: Record<string, unknown>;
}
/**
* Serve until stdin closes, which is how a host ends a session.
*
* The tool list is fetched once, on the first `tools/list`, and kept. An agent asks for it repeatedly
* and the catalogue's answer does not change mid-session; refetching would make every turn cost a
* round trip to a module for something nobody changed.
*/
export async function serveMcp(bus: Broker, who: string): Promise<void> {
let known: Tool[] | undefined;
const say = (message: unknown) => {
process.stdout.write(`${JSON.stringify(message)}\n`);
};
const answer = (id: Request["id"], result: unknown) => say({ jsonrpc: "2.0", id, result });
const refuse = (id: Request["id"], code: number, message: string) =>
say({ jsonrpc: "2.0", id, error: { code, message } });
for await (const line of lines()) {
let request: Request;
try {
request = JSON.parse(line) as Request;
} catch {
// Unparseable, and with no id there is nobody to tell. Skipped rather than answered, because a
// reply to a request that was never framed is noise on the same channel.
continue;
}
// A notification has no id and expects no answer; `initialized` is the one every host sends.
const notification = request.id === undefined || request.id === null;
switch (request.method) {
case "initialize":
answer(request.id, {
protocolVersion: PROTOCOL,
capabilities: { tools: {} },
serverInfo: { name: "mesh", version: "1" },
// Said in the handshake, because an agent that knows whose authority it is acting under can
// say so when a call is refused — and a refusal is the one thing here that is not the
// mesh's fault or the tool's.
instructions:
`These are the tools of a Novox mesh, reached as ${who}. Every call goes to the module ` +
`that serves it; what may be called was fixed when this credential was issued, so a ` +
`refusal means the credential, not the tool.`,
});
break;
case "notifications/initialized":
break;
case "tools/list": {
try {
known ??= await toolsOn(bus);
} catch (e) {
refuse(request.id, -32603, whyItFailed("mesh-catalog.catalog_tools", e));
break;
}
answer(request.id, {
tools: known.map((t) => ({
name: `${t.module}.${t.name}`,
description: t.description ?? `${t.name}, served by ${t.module}`,
// The module's own schema, passed through. An empty object is a tool that takes nothing,
// which is a real answer and not a missing one.
inputSchema: t.input ?? { type: "object", properties: {} },
})),
});
break;
}
case "tools/call": {
const name = String(request.params?.name ?? "");
const args = request.params?.arguments ?? {};
try {
const result = await callTool(bus, name, args);
// Text, because that is what every host renders. The content is the module's answer as
// JSON, unshaped: an adapter that flattened it would be deciding what matters in somebody
// else's answer.
answer(request.id, {
content: [{ type: "text", text: JSON.stringify(result, null, 2) }],
});
} catch (e) {
// **An error the agent can act on, not a stack.** isError rather than a protocol failure,
// because the call was well-formed and the mesh answered it — with a refusal, an absence or
// a fault, and the words say which.
answer(request.id, {
content: [{ type: "text", text: whyItFailed(name, e) }],
isError: true,
});
}
break;
}
default:
if (!notification) {
refuse(request.id, -32601, `mesh's MCP surface has no ${request.method}`);
}
}
}
}
/** stdin as newline-framed messages, which is what MCP over stdio is. */
async function* lines(): AsyncGenerator<string> {
let buffered = "";
for await (const chunk of process.stdin) {
buffered += (chunk as Buffer).toString("utf8");
let at: number;
while ((at = buffered.indexOf("\n")) >= 0) {
const line = buffered.slice(0, at).trim();
buffered = buffered.slice(at + 1);
if (line !== "") yield line;
}
}
if (buffered.trim() !== "") yield buffered.trim();
}
+145
View File
@@ -0,0 +1,145 @@
#!/usr/bin/env node
/**
* `mesh` — the mesh's tools from a workstation, for a person (novox/hq design 25 §7).
*
* Three verbs and nothing else. What tools are there, call one, and serve the same two to an agent
* over MCP. Deliberately thin: everything that could be a decision is one the mesh already made, and a
* client that grew opinions would be a second place the mesh's behaviour is defined.
*
* mesh tools what this credential may call
* mesh call <module>.<tool> [json] call one, arguments as JSON on the command line or on stdin
* mesh mcp the same, as an MCP server over stdio
*
* The credential comes from MESH_CREDENTIAL, or --credential. It is the JSON `operator issue` printed.
*/
import { readFile } from "node:fs/promises";
import { callTool, connectAs, credentialFrom, toolsOn, whyItFailed, type Tool } from "./client.js";
import { serveMcp } from "./mcp.js";
const usage = `mesh tools
mesh call <module>.<tool> [json]
mesh mcp
--credential <file> the JSON \`operator issue\` printed; default $MESH_CREDENTIAL`;
async function main(argv: string[]): Promise<number> {
const args = [...argv];
let credentialPath = process.env.MESH_CREDENTIAL ?? "";
for (let i = 0; i < args.length; i++) {
if (args[i] === "--credential") {
credentialPath = args[i + 1] ?? "";
args.splice(i, 2);
i--;
}
}
const verb = args.shift();
if (!verb || verb === "help" || verb === "--help") {
console.log(usage);
return verb ? 0 : 1;
}
if (!credentialPath) {
console.error(
"no credential: set MESH_CREDENTIAL or pass --credential <file>. It is the JSON " +
"`operator issue` printed, saved verbatim.",
);
return 1;
}
const held = await credentialFrom(credentialPath);
const bus = await connectAs(held);
try {
switch (verb) {
case "tools":
return await listing(bus, held.person);
case "call":
return await calling(bus, args);
case "mcp":
// Serves until stdin closes, which is how an MCP host ends a session.
await serveMcp(bus, held.person ?? held.user ?? "somebody");
return 0;
default:
console.error(`mesh has no "${verb}".\n\n${usage}`);
return 1;
}
} finally {
await bus.close();
}
}
async function listing(bus: Awaited<ReturnType<typeof connectAs>>, who?: string): Promise<number> {
let tools: Tool[];
try {
tools = await toolsOn(bus);
} catch (e) {
console.error(whyItFailed("mesh-catalog.catalog_tools", e));
return 1;
}
if (tools.length === 0) {
console.log("the catalogue lists no tools; nothing on this mesh serves any");
return 0;
}
// **What the catalogue has, not what this credential may call.** The two differ and the difference
// is the point: a person seeing only their own tools cannot tell "not installed" from "not yours",
// and those need different people to fix them.
for (const t of tools) {
const name = `${t.module}.${t.name}`;
console.log(t.description ? `${name.padEnd(36)} ${t.description}` : name);
}
if (who) {
console.log(`\nthis is what the mesh has. What ${who} may call was fixed when the credential was issued.`);
}
return 0;
}
async function calling(
bus: Awaited<ReturnType<typeof connectAs>>,
args: string[],
): Promise<number> {
const key = args.shift();
if (!key) {
console.error("mesh call <module>.<tool> [json]");
return 1;
}
const raw = args.length > 0 ? args.join(" ") : await maybeStdin();
let parsed: unknown = {};
if (raw.trim() !== "") {
try {
parsed = JSON.parse(raw);
} catch (e) {
console.error(`the arguments are not JSON: ${(e as Error).message}`);
return 1;
}
}
try {
const answer = await callTool(bus, key, parsed);
console.log(JSON.stringify(answer, null, 2));
return 0;
} catch (e) {
console.error(whyItFailed(key, e));
return 1;
}
}
/** Arguments on stdin, for a call whose JSON is too long or too quoted to type. Empty when stdin is a
* terminal, so `mesh call x.y` with no arguments does not hang waiting for something nobody is
* typing. */
async function maybeStdin(): Promise<string> {
if (process.stdin.isTTY) return "";
const chunks: Buffer[] = [];
for await (const chunk of process.stdin) chunks.push(chunk as Buffer);
return Buffer.concat(chunks).toString("utf8");
}
// Only when run, so a test can import the pieces.
if (process.argv[1] && import.meta.url === new URL(`file://${process.argv[1]}`).href) {
main(process.argv.slice(2))
.then((code) => process.exit(code))
.catch((e) => {
console.error(e instanceof Error ? e.message : String(e));
process.exit(1);
});
}
export { main, usage };
export const _readFile = readFile;
+105
View File
@@ -0,0 +1,105 @@
/**
* A person's client, against a real bus.
*
* What is worth checking is not that a request/reply works — the runtime's own tests cover that — but
* that the two surfaces are the same thing. An agent and a person must see the same tools and get the
* same answers, or the MCP surface becomes a second definition of what a tool is.
*
* docker run -d --rm --name t -p 14232:4222 nats:2.10-alpine -js
* MESH_TEST_NATS=nats://127.0.0.1:14232 node --test --experimental-strip-types test/client.test.ts
*/
import assert from "node:assert/strict";
import { test } from "node:test";
// The built output, not the source: the client imports its siblings as `.js`, which is what ships and
// what every other file here does, and cannot be loaded as TypeScript directly. `pretest` builds.
import { connectNats } from "../dist/broker-nats.js";
import { callTool, toolsOn, whyItFailed } from "../dist/client.js";
const url = process.env.MESH_TEST_NATS;
/** A module serving the catalogue's tool list and one tool of its own, so the client has a mesh to
* talk to. Two connections, because a person and a module are different users even in a test. */
async function aMeshWithTools() {
const catalogue = await connectNats({ url: url!, module: "mesh-catalog" });
const shop = await connectNats({ url: url!, module: "shop" });
await catalogue.handle("catalog_tools", async () => ({
tools: [
{ module: "shop", name: "price", description: "what something costs", input: { type: "object" } },
{ module: "mesh-catalog", name: "catalog_tools", description: "what tools the mesh has" },
],
}));
await shop.handle("price", async (body: { of?: string }) => ({ of: body.of ?? "nothing", cost: 12 }));
return {
async close() {
await catalogue.close();
await shop.close();
},
};
}
test("a person sees what the catalogue says the mesh has, sorted", async (t) => {
if (!url) return t.skip("MESH_TEST_NATS unset");
const mesh = await aMeshWithTools();
const person = await connectNats({ url, module: "person.ada" });
try {
const tools = await toolsOn(person);
assert.deepEqual(
tools.map((x) => `${x.module}.${x.name}`),
["mesh-catalog.catalog_tools", "shop.price"],
"the list is what the catalogue answered, in a stable order",
);
} finally {
await person.close();
await mesh.close();
}
});
test("a person calls a tool and gets the module's own answer, unshaped", async (t) => {
if (!url) return t.skip("MESH_TEST_NATS unset");
const mesh = await aMeshWithTools();
const person = await connectNats({ url, module: "person.ada" });
try {
const answer = await callTool(person, "shop.price", { of: "a hat" });
assert.deepEqual(answer, { of: "a hat", cost: 12 });
} finally {
await person.close();
await mesh.close();
}
});
test("a tool nobody serves says so at once, and says what to do about it", async (t) => {
if (!url) return t.skip("MESH_TEST_NATS unset");
const person = await connectNats({ url: url!, module: "person.ada" });
try {
const began = Date.now();
await assert.rejects(() => callTool(person, "ghost.missing", {}));
// At once, not after the whole wait: "that module is down" and "that tool is slow" need
// different things done, and a timeout cannot tell them apart.
assert.ok(Date.now() - began < 5_000, "a tool nobody serves waited out the timeout");
} finally {
await person.close();
}
});
test("a name that is not <module>.<tool> is refused before anything is sent", async (t) => {
if (!url) return t.skip("MESH_TEST_NATS unset");
const person = await connectNats({ url: url!, module: "person.ada" });
try {
await assert.rejects(() => callTool(person, "price", {}), /does not name a tool/);
} finally {
await person.close();
}
});
test("each way a call fails says what to do about it", () => {
// The three answers a person actually gets. Without this they are one timeout and a stack trace,
// and the remedies are in three different places.
assert.match(whyItFailed("shop.price", new Error("no responders")), /nothing serves shop\.price/);
assert.match(
whyItFailed("shop.price", new Error("Permissions Violation for Publish")),
/may not call shop\.price/,
);
assert.match(whyItFailed("shop.price", new Error("timeout")), /did not answer in time/);
assert.match(whyItFailed("shop.price", new Error("something else")), /something else/);
});
+124
View File
@@ -0,0 +1,124 @@
/**
* The MCP surface, driven the way a host drives it.
*
* **The claim worth checking is that it is the same thing the command line is.** An agent and a
* person must see the same tools and get the same answers, or this becomes a second definition of what
* a tool is — which is exactly what a thin adapter is supposed to avoid.
*
* docker run -d --rm --name t -p 14232:4222 nats:2.10-alpine -js
* MESH_TEST_NATS=nats://127.0.0.1:14232 node --test --experimental-strip-types test/mcp.test.ts
*/
import assert from "node:assert/strict";
import { test } from "node:test";
import { spawn } from "node:child_process";
import { connectNats } from "../dist/broker-nats.js";
const url = process.env.MESH_TEST_NATS;
/** A module answering the catalogue's list and one tool, plus a credential file the client reads. */
async function aMeshAndACredential(t: { after: (fn: () => Promise<void> | void) => void }) {
const catalogue = await connectNats({ url: url!, module: "mesh-catalog" });
const shop = await connectNats({ url: url!, module: "shop" });
await catalogue.handle("catalog_tools", async () => ({
tools: [{ module: "shop", name: "price", description: "what something costs" }],
}));
await shop.handle("price", async (body: { of?: string }) => ({ of: body.of ?? "nothing", cost: 12 }));
t.after(async () => {
await catalogue.close();
await shop.close();
});
const { mkdtemp, writeFile } = await import("node:fs/promises");
const { join } = await import("node:path");
const dir = await mkdtemp("/tmp/mesh-client-");
const path = join(dir, "credential.json");
await writeFile(
path,
JSON.stringify({ url, user: "person.ada", password: "x", person: "ada", invokes: ["shop.price"] }),
);
return path;
}
/** Drive `mesh mcp` over stdio and collect the replies, as a host would. */
function driving(credential: string, requests: unknown[]): Promise<Record<string, any>[]> {
return new Promise((resolve, reject) => {
const child = spawn(process.execPath, ["dist/mesh.js", "mcp", "--credential", credential], {
stdio: ["pipe", "pipe", "pipe"],
});
let out = "";
let err = "";
child.stdout.on("data", (d) => (out += d.toString()));
child.stderr.on("data", (d) => (err += d.toString()));
child.on("error", reject);
child.on("close", () => {
const replies = out
.split("\n")
.filter((l) => l.trim() !== "")
.map((l) => JSON.parse(l) as Record<string, any>);
if (replies.length === 0 && err !== "") reject(new Error(err));
else resolve(replies);
});
for (const r of requests) child.stdin.write(`${JSON.stringify(r)}\n`);
child.stdin.end();
});
}
test("a host initialises, lists the mesh's tools and calls one", async (t) => {
if (!url) return t.skip("MESH_TEST_NATS unset");
const credential = await aMeshAndACredential(t);
const replies = await driving(credential, [
{ jsonrpc: "2.0", id: 1, method: "initialize", params: {} },
{ jsonrpc: "2.0", method: "notifications/initialized" },
{ jsonrpc: "2.0", id: 2, method: "tools/list" },
{ jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "shop.price", arguments: { of: "a hat" } } },
]);
const byId = new Map(replies.map((r) => [r.id, r]));
// A notification is answered with nothing, or a host waiting on ids sees a reply it cannot match.
assert.equal(replies.length, 3, `expected three replies, got ${JSON.stringify(replies)}`);
const hello = byId.get(1)!.result;
assert.equal(hello.protocolVersion, "2024-11-05");
assert.ok(hello.capabilities.tools, "a server offering no tools is not this one");
assert.match(hello.instructions, /ada/, "the handshake says whose authority a call is made under");
const listed = byId.get(2)!.result.tools;
assert.equal(listed.length, 1);
assert.equal(listed[0].name, "shop.price", "a tool is named the way a person names it");
assert.ok(listed[0].inputSchema, "a tool with no schema is one an agent cannot call");
const called = byId.get(3)!.result;
assert.ok(!called.isError, `the call failed: ${JSON.stringify(called)}`);
// The module's own answer, unshaped. An adapter that summarised it would be deciding what matters
// in somebody else's answer.
assert.deepEqual(JSON.parse(called.content[0].text), { of: "a hat", cost: 12 });
});
test("a tool nobody serves comes back as an error the agent can act on", async (t) => {
if (!url) return t.skip("MESH_TEST_NATS unset");
const credential = await aMeshAndACredential(t);
const replies = await driving(credential, [
{ jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "ghost.missing", arguments: {} } },
]);
const result = replies[0].result;
// isError, not a protocol failure: the call was well-formed and the mesh answered it — with an
// absence. A JSON-RPC error would tell the agent its request was malformed, which it was not.
assert.ok(result?.isError, `expected a tool error, got ${JSON.stringify(replies[0])}`);
assert.match(result.content[0].text, /nothing serves ghost\.missing/);
});
test("a method this surface does not have is refused, and a notification is not", async (t) => {
if (!url) return t.skip("MESH_TEST_NATS unset");
const credential = await aMeshAndACredential(t);
const replies = await driving(credential, [
{ jsonrpc: "2.0", id: 1, method: "resources/list" },
{ jsonrpc: "2.0", method: "notifications/cancelled" },
]);
assert.equal(replies.length, 1, "a notification was answered");
assert.equal(replies[0].error.code, -32601);
assert.match(replies[0].error.message, /resources\/list/);
});